OPENDJ-1053 Improve logging of replication load balancing and fail-over in ...

Jean-Noel Rouvignac

06.51.2013 b161a27ed6c0b7d72b241e9990f9917c1ca9c0ac

OPENDJ-1053 Improve logging of replication load balancing and fail-over in order to diagnose the causes of these events



Functional tests found a ConcurrentModificationException with r9753 (see at the end of the commit message):
The code was removing elements from a Collection while it was iterating on the exact same Collection.



ReplicationBroker.java:
In inner class LocalEvaluation:
- Renamed "filteredRSs" to "accepted".
- Added getAcceptedRSInfos() to allow iterating on them without triggering ConcurrentModificationException.
In rejectAll*(), called getAcceptedRSInfos().
In getCSN(), renamed parameters.

Fixed ConcurrentModificationException.

ComputeBestServerTest.java:
Added a test2ServersUpToDateAnd1EvenMoreUpToDate() to exhibit the ConcurrentModificationException.
Extracted method newServerState() and used it in all tests.
In all test methods but one, do not update the ServerState with unused CSNs because they make the tests harder to read. Did not change test3Servers() because I believe it is good to keep one test testing the unneded CSNs do not impact the test results.




java.util.ConcurrentModificationException
	at java.util.HashMap$HashIterator.nextEntry(HashMap.java:793)
	at java.util.HashMap$ValueIterator.next(HashMap.java:822)
	at org.opends.server.replication.service.ReplicationBroker.rejectAllWithRSOnDifferentVMThanDS(ReplicationBroker.java:1908)
	at org.opends.server.replication.service.ReplicationBroker.filterServersOnSameHost(ReplicationBroker.java:1878)
	at org.opends.server.replication.service.ReplicationBroker.computeBestReplicationServer(ReplicationBroker.java:1600)
	at org.opends.server.replication.service.ReplicationBroker.connectAsDataServer(ReplicationBroker.java:820)
	at org.opends.server.replication.service.ReplicationBroker.connect(ReplicationBroker.java:691)
	at org.opends.server.replication.service.ReplicationBroker.start(ReplicationBroker.java:221)
	at org.opends.server.replication.service.ReplicationDomain.enableService(ReplicationDomain.java:3024)
	at org.opends.server.replication.service.ReplicationDomain.changeConfig(ReplicationDomain.java:3054)
	at org.opends.server.replication.plugin.LDAPReplicationDomain.applyConfigurationChange(LDAPReplicationDomain.java:4089)
	at org.opends.server.replication.plugin.LDAPReplicationDomain.applyConfigurationChange(LDAPReplicationDomain.java:96)
	at org.opends.server.admin.server.ServerManagedObjectChangeListenerAdaptor.applyConfigurationChange(ServerManagedObjectChangeListenerAdaptor.java:74)
	at org.opends.server.admin.server.ConfigChangeListenerAdaptor.applyConfigurationChange(ConfigChangeListenerAdaptor.java:342)
	at org.opends.server.extensions.ConfigFileHandler.replaceEntry(ConfigFileHandler.java:1610)
	at org.opends.server.workflowelement.localbackend.LocalBackendModifyOperation.processModify(LocalBackendModifyOperation.java:599)
	at org.opends.server.workflowelement.localbackend.LocalBackendModifyOperation.processLocalModify(LocalBackendModifyOperation.java:299)
	at org.opends.server.workflowelement.localbackend.LocalBackendWorkflowElement.execute(LocalBackendWorkflowElement.java:690)
	at org.opends.server.core.WorkflowImpl.execute(WorkflowImpl.java:197)
	at org.opends.server.core.WorkflowTopologyNode.execute(WorkflowTopologyNode.java:100)
	at org.opends.server.core.ModifyOperationBasis.run(ModifyOperationBasis.java:451)
	at org.opends.server.core.SynchronousStrategy.enqueueRequest(SynchronousStrategy.java:49)
	at org.opends.server.protocols.ldap.LDAPClientConnection.addOperationInProgress(LDAPClientConnection.java:1276)
	at org.opends.server.protocols.ldap.LDAPClientConnection.processModifyRequest(LDAPClientConnection.java:2268)
	at org.opends.server.protocols.ldap.LDAPClientConnection.processLDAPMessage(LDAPClientConnection.java:1745)
	at org.opends.server.protocols.ldap.LDAPRequestHandler.run(LDAPRequestHandler.java:194)

 opends/src/server/org/opends/server/replication/service/ReplicationBroker.java

@@ -1487,26 +1487,33 @@
   */
  private static class LocalEvaluation
  {
    private final Map<Integer, ReplicationServerInfo> filteredRSs =
    private final Map<Integer, ReplicationServerInfo> accepted =
        new HashMap<Integer, ReplicationServerInfo>();
    private final Map<ReplicationServerInfo, Message> rsEvals =
        new HashMap<ReplicationServerInfo, Message>();

    private void accept(Integer rsId, ReplicationServerInfo rsInfo)
    {
      this.rsEvals.remove(rsInfo); // undo reject
      this.filteredRSs.put(rsId, rsInfo);
      // forget previous eval, including undoing reject
      this.rsEvals.remove(rsInfo);
      this.accepted.put(rsId, rsInfo);
    }

    private void reject(ReplicationServerInfo rsInfo, Message reason)
    {
      this.filteredRSs.remove(rsInfo.getServerId()); // undo accept
      this.accepted.remove(rsInfo.getServerId()); // undo accept
      this.rsEvals.put(rsInfo, reason);
    }

    private Map<Integer, ReplicationServerInfo> getAccepted()
    {
      return filteredRSs;
      return accepted;
    }

    private ReplicationServerInfo[] getAcceptedRSInfos()
    {
      return accepted.values().toArray(
          new ReplicationServerInfo[accepted.size()]);
    }

    public Map<Integer, Message> getRejected()
@@ -1521,7 +1528,7 @@

    private boolean hasAcceptedAny()
    {
      return !filteredRSs.isEmpty();
      return !accepted.isEmpty();
    }

  }
@@ -1818,20 +1825,20 @@
    evals.keepBest(mostUpToDateEval);
  }

  private static CSN getCSN(ServerState localState, int localServerId)
  private static CSN getCSN(ServerState state, int serverId)
  {
    final CSN csn = localState.getCSN(localServerId);
    final CSN csn = state.getCSN(serverId);
    if (csn != null)
    {
      return csn;
    }
    return new CSN(0, 0, localServerId);
    return new CSN(0, 0, serverId);
  }

  private static void rejectAllWithRSIsLaterThanBestRS(
      final LocalEvaluation eval, int localServerId, CSN localCSN)
  {
    for (ReplicationServerInfo rsInfo : eval.getAccepted().values())
    for (ReplicationServerInfo rsInfo : eval.getAcceptedRSInfos())
    {
      final String rsCSN =
          getCSN(rsInfo.getServerState(), localServerId).toStringUI();
@@ -1902,7 +1909,7 @@
  private static void rejectAllWithRSOnDifferentVMThanDS(LocalEvaluation eval,
      int localServerId)
  {
    for (ReplicationServerInfo rsInfo : eval.getAccepted().values())
    for (ReplicationServerInfo rsInfo : eval.getAcceptedRSInfos())
    {
      eval.reject(rsInfo, NOTE_RS_ON_DIFFERENT_VM_THAN_DS.get(
          rsInfo.getServerId(), localServerId));

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

@@ -87,6 +87,16 @@
    }
  }

  private ServerState newServerState(CSN... csns)
  {
    ServerState result = new ServerState();
    for (CSN csn : csns)
    {
      result.update(csn);
    }
    return result;
  }

  /**
   * Test with one replication server, nobody has a CSN (simulates) very first
   * connection.
@@ -95,18 +105,10 @@
  public void testNullCSNBoth() throws Exception
  {
    String testCase = "testNullCSNBoth";

    debugInfo("Starting " + testCase);

    // Create my state
    ServerState mySt = new ServerState();
    mySt.update(new CSN(2, 0, myId2)); // Should not be used inside algo
    mySt.update(new CSN(3, 0, myId3)); // Should not be used inside algo

    // State for server 1
    ServerState aState = new ServerState();
    aState.update(new CSN(0, 0, myId2));
    aState.update(new CSN(0, 0, myId3));
    ServerState mySt = newServerState();
    ServerState aState = newServerState();

    Map<Integer, ReplicationServerInfo> rsInfos =
        newRSInfos(newRSInfo(11, WINNER, aState, 0, 1));
@@ -145,19 +147,10 @@
  public void testNullCSNDS() throws Exception
  {
    String testCase = "testNullCSNDS";

    debugInfo("Starting " + testCase);

    // Create my state
    ServerState mySt = new ServerState();
    mySt.update(new CSN(2, 0, myId2));// Should not be used inside algo
    mySt.update(new CSN(3, 0, myId3));// Should not be used inside algo

    // State for server 1
    ServerState aState = new ServerState();
    aState.update(new CSN(0, 0, myId1));
    aState.update(new CSN(0, 0, myId2));
    aState.update(new CSN(0, 0, myId3));
    ServerState mySt = newServerState();
    ServerState aState = newServerState(new CSN(0, 0, myId1));

    Map<Integer, ReplicationServerInfo> rsInfos =
        newRSInfos(newRSInfo(11, WINNER, aState, 0, 1));
@@ -177,19 +170,10 @@
  public void testNullCSNRS() throws Exception
  {
    String testCase = "testNullCSNRS";

    debugInfo("Starting " + testCase);

    // Create my state
    ServerState mySt = new ServerState();
    mySt.update(new CSN(1, 0, myId1));
    mySt.update(new CSN(2, 0, myId2)); // Should not be used inside algo
    mySt.update(new CSN(3, 0, myId3)); // Should not be used inside algo

    // State for server 1
    ServerState aState = new ServerState();
    aState.update(new CSN(0, 0, myId2));
    aState.update(new CSN(0, 0, myId3));
    ServerState mySt = newServerState(new CSN(1, 0, myId1));
    ServerState aState = newServerState();

    Map<Integer, ReplicationServerInfo> rsInfos =
        newRSInfos(newRSInfo(11, WINNER, aState, 0, 1));
@@ -208,20 +192,10 @@
  public void test1ServerUp() throws Exception
  {
    String testCase = "test1ServerUp";

    debugInfo("Starting " + testCase);

    // Create my state
    ServerState mySt = new ServerState();
    mySt.update(new CSN(1, 0, myId1));
    mySt.update(new CSN(2, 0, myId2)); // Should not be used inside algo
    mySt.update(new CSN(3, 0, myId3)); // Should not be used inside algo

    // State for server 1
    ServerState aState = new ServerState();
    aState.update(new CSN(1, 0, myId1));
    aState.update(new CSN(1, 0, myId2));
    aState.update(new CSN(1, 0, myId3));
    ServerState mySt = newServerState(new CSN(1, 0, myId1));
    ServerState aState = newServerState(new CSN(1, 0, myId1));

    Map<Integer, ReplicationServerInfo> rsInfos =
        newRSInfos(newRSInfo(11, WINNER, aState, 0, 1));
@@ -240,26 +214,11 @@
  public void test2ServersUp() throws Exception
  {
    String testCase = "test2ServersUp";

    debugInfo("Starting " + testCase);

    // Create my state
    ServerState mySt = new ServerState();
    mySt.update(new CSN(1, 0, myId1));
    mySt.update(new CSN(2, 0, myId2)); // Should not be used inside algo
    mySt.update(new CSN(3, 0, myId3)); // Should not be used inside algo

    // State for server 1
    ServerState aState1 = new ServerState();
    aState1.update(new CSN(1, 0, myId1));
    aState1.update(new CSN(1, 0, myId2));
    aState1.update(new CSN(1, 0, myId3));

    // State for server 2
    ServerState aState2 = new ServerState();
    aState2.update(new CSN(2, 0, myId1));
    aState2.update(new CSN(1, 0, myId2));
    aState2.update(new CSN(1, 0, myId3));
    ServerState mySt = newServerState(new CSN(1, 0, myId1));
    ServerState aState1 = newServerState(new CSN(1, 0, myId1));
    ServerState aState2 = newServerState(new CSN(2, 0, myId1));

    Map<Integer, ReplicationServerInfo> rsInfos = newRSInfos(
        newRSInfo(11, LOOSER1, aState1, 0, 1),
@@ -281,30 +240,15 @@
  public void testDiffGroup2ServersUp() throws Exception
  {
    String testCase = "testDiffGroup2ServersUp";

    debugInfo("Starting " + testCase);

    // Create my state
    ServerState mySt = new ServerState();
    mySt.update(new CSN(1, 0, myId1));
    mySt.update(new CSN(2, 0, myId2)); // Should not be used inside algo
    mySt.update(new CSN(3, 0, myId3)); // Should not be used inside algo

    // State for server 1
    ServerState aState1 = new ServerState();
    aState1.update(new CSN(1, 0, myId1));
    aState1.update(new CSN(1, 0, myId2));
    aState1.update(new CSN(1, 0, myId3));
    // This server has less changes than the other one but it has the same
    // group id as us so he should be the winner

    // State for server 2
    ServerState aState2 = new ServerState();
    aState2.update(new CSN(2, 0, myId1));
    aState2.update(new CSN(1, 0, myId2));
    aState2.update(new CSN(1, 0, myId3));
    ServerState mySt = newServerState(new CSN(1, 0, myId1));
    ServerState aState1 = newServerState(new CSN(1, 0, myId1));
    ServerState aState2 = newServerState(new CSN(2, 0, myId1));

    Map<Integer, ReplicationServerInfo> rsInfos = newRSInfos(
        // This server has less changes than the other one but it has the same
        // group id as us so he should be the winner
        newRSInfo(11, WINNER, aState1, 0, 1),
        newRSInfo(12, LOOSER1, aState2, 0, 2));
    RSEvaluations evals =
@@ -371,26 +315,11 @@
  public void testNotOurGroup() throws Exception
  {
    String testCase = "testNotOurGroup";

    debugInfo("Starting " + testCase);

    // Create my state
    ServerState mySt = new ServerState();
    mySt.update(new CSN(1, 0, myId1));
    mySt.update(new CSN(2, 0, myId2)); // Should not be used inside algo
    mySt.update(new CSN(3, 0, myId3)); // Should not be used inside algo

    // State for server 1
    ServerState aState1 = new ServerState();
    aState1.update(new CSN(1, 0, myId1));
    aState1.update(new CSN(1, 0, myId2));
    aState1.update(new CSN(1, 0, myId3));

    // State for server 2
    ServerState aState2 = new ServerState();
    aState2.update(new CSN(2, 0, myId1));
    aState2.update(new CSN(2, 0, myId2));
    aState2.update(new CSN(2, 0, myId3));
    ServerState mySt = newServerState(new CSN(1, 0, myId1));
    ServerState aState1 = newServerState(new CSN(1, 0, myId1));
    ServerState aState2 = newServerState(new CSN(2, 0, myId1));

    Map<Integer, ReplicationServerInfo> rsInfos = newRSInfos(
        newRSInfo(11, LOOSER1, aState1, 0, 2),
@@ -412,32 +341,42 @@
  public void test3ServersUp() throws Exception
  {
    String testCase = "test3ServersUp";

    debugInfo("Starting " + testCase);

    // Create my state
    ServerState mySt = new ServerState();
    mySt.update(new CSN(1, 0, myId1));
    mySt.update(new CSN(2, 0, myId2)); // Should not be used inside algo
    mySt.update(new CSN(3, 0, myId3)); // Should not be used inside algo
    ServerState mySt = newServerState(new CSN(1, 0, myId1));
    ServerState aState1 = newServerState(new CSN(1, 0, myId1));
    ServerState aState2 = newServerState(new CSN(2, 0, myId1));
    ServerState aState3 = newServerState(new CSN(3, 0, myId1));

    // State for server 1
    ServerState aState1 = new ServerState();
    aState1.update(new CSN(1, 0, myId1));
    aState1.update(new CSN(1, 0, myId2));
    aState1.update(new CSN(1, 0, myId3));
    Map<Integer, ReplicationServerInfo> rsInfos = newRSInfos(
        newRSInfo(11, LOOSER1, aState1, 0, 1),
        newRSInfo(12, LOOSER2, aState2, 0, 1),
        newRSInfo(13, WINNER, aState3, 0, 1));
    RSEvaluations evals =
      computeBestReplicationServer(true, -1, mySt, rsInfos, myId1, (byte)1, 0);

    // State for server 2
    ServerState aState2 = new ServerState();
    aState2.update(new CSN(2, 0, myId1));
    aState2.update(new CSN(1, 0, myId2));
    aState2.update(new CSN(4, 0, myId3));
    assertEquals(evals.getBestRS().getServerURL(), WINNER,
        "Wrong best replication server.");
    containsOnly(evals.getEvaluations(),
        entry(11, NOTE_RS_LATER_THAN_ANOTHER_RS_MORE_UP_TO_DATE_THAN_LOCAL_DS),
        entry(12, NOTE_RS_LATER_THAN_ANOTHER_RS_MORE_UP_TO_DATE_THAN_LOCAL_DS),
        entry(13, NOTE_BEST_RS));
  }

    // State for server 3
    ServerState aState3 = new ServerState();
    aState3.update(new CSN(3, 0, myId1));
    aState3.update(new CSN(2, 0, myId2));
    aState3.update(new CSN(1, 0, myId3));
  /**
   * Test with 3 replication servers: 2 are up to date with the directory
   * server, 1 is more up to date than the directory server.
   */
  @Test
  public void test2ServersUpToDateAnd1EvenMoreUpToDate() throws Exception
  {
    String testCase = "test3ServersUp";
    debugInfo("Starting " + testCase);

    ServerState mySt = newServerState(new CSN(1, 0, myId1));
    ServerState aState1 = newServerState(new CSN(1, 0, myId1));
    ServerState aState2 = newServerState(new CSN(1, 0, myId1));
    ServerState aState3 = newServerState(new CSN(2, 0, myId1));

    Map<Integer, ReplicationServerInfo> rsInfos = newRSInfos(
        newRSInfo(11, LOOSER1, aState1, 0, 1),
@@ -461,32 +400,12 @@
  public void testDiffGroup3ServersUp() throws Exception
  {
    String testCase = "testDiffGroup3ServersUp";

    debugInfo("Starting " + testCase);

    // Create my state
    ServerState mySt = new ServerState();
    mySt.update(new CSN(1, 0, myId1));
    mySt.update(new CSN(2, 0, myId2)); // Should not be used inside algo
    mySt.update(new CSN(3, 0, myId3)); // Should not be used inside algo

    // State for server 1
    ServerState aState1 = new ServerState();
    aState1.update(new CSN(1, 0, myId1));
    aState1.update(new CSN(1, 0, myId2));
    aState1.update(new CSN(1, 0, myId3));

    // State for server 2
    ServerState aState2 = new ServerState();
    aState2.update(new CSN(2, 0, myId1));
    aState2.update(new CSN(1, 0, myId2));
    aState2.update(new CSN(3, 0, myId3));

    // State for server 3
    ServerState aState3 = new ServerState();
    aState3.update(new CSN(3, 0, myId1));
    aState3.update(new CSN(2, 0, myId2));
    aState3.update(new CSN(1, 0, myId3));
    ServerState mySt = newServerState(new CSN(1, 0, myId1));
    ServerState aState1 = newServerState(new CSN(1, 0, myId1));
    ServerState aState2 = newServerState(new CSN(2, 0, myId1));
    ServerState aState3 = newServerState(new CSN(3, 0, myId1));

    Map<Integer, ReplicationServerInfo> rsInfos = newRSInfos(
        newRSInfo(11, LOOSER1, aState1, 0, 1),
@@ -512,20 +431,10 @@
  public void test1ServerLate() throws Exception
  {
    String testCase = "test1ServerLate";

    debugInfo("Starting " + testCase);

    // Create my state
    ServerState mySt = new ServerState();
    mySt.update(new CSN(1, 0, myId1));
    mySt.update(new CSN(2, 0, myId2));// Should not be used inside algo
    mySt.update(new CSN(3, 0, myId3));// Should not be used inside algo

    // State for server 1
    ServerState aState = new ServerState();
    aState.update(new CSN(0, 0, myId1));
    aState.update(new CSN(1, 0, myId2));
    aState.update(new CSN(1, 0, myId3));
    ServerState mySt = newServerState(new CSN(1, 0, myId1));
    ServerState aState = newServerState(new CSN(0, 0, myId1));

    Map<Integer, ReplicationServerInfo> rsInfos =
        newRSInfos(newRSInfo(11, WINNER, aState, 0, 1));
@@ -597,7 +506,6 @@
      throws Exception
  {
    String testCase = "test3ServersLate";

    debugInfo("Starting " + testCase);

    // definitions for server names
@@ -606,32 +514,32 @@
    final String LOOSER2 = "localhost:789";

    // Create my state
    ServerState mySt = new ServerState();
    mySt.update(new CSN(4, 0, myId1));
    mySt.update(new CSN(2, 0, myId2));// Should not be used inside algo
    mySt.update(new CSN(3, 0, myId3));// Should not be used inside algo
    ServerState mySt = newServerState(
        new CSN(4, 0, myId1),
        new CSN(2, 0, myId2),  // myId2 should not be used inside algo (same for the other ServerStates)
        new CSN(3, 0, myId3)); // myId3 should not be used inside algo (same for the other ServerStates)

    // State for server 1
    ServerState aState1 = new ServerState();
    aState1.update(new CSN(looser1T1, 0, myId1));
    aState1.update(new CSN(looser1T2, 0, myId2));
    aState1.update(new CSN(looser1T3, 0, myId3));
    ServerState aState1 = newServerState(
        new CSN(looser1T1, 0, myId1),
        new CSN(looser1T2, 0, myId2),
        new CSN(looser1T3, 0, myId3));
    if (looser1IsLocal)
      ReplicationServer.onlyForTestsAddlocalReplicationServer(LOOSER1);

    // State for server 2
    ServerState aState2 = new ServerState();
    aState2.update(new CSN(winnerT1, 0, myId1));
    aState2.update(new CSN(winnerT2, 0, myId2));
    aState2.update(new CSN(winnerT3, 0, myId3));
    ServerState aState2 = newServerState(
        new CSN(winnerT1, 0, myId1),
        new CSN(winnerT2, 0, myId2),
        new CSN(winnerT3, 0, myId3));
    if (winnerIsLocal)
      ReplicationServer.onlyForTestsAddlocalReplicationServer(WINNER);

    // State for server 3
    ServerState aState3 = new ServerState();
    aState3.update(new CSN(looser2T1, 0, myId1));
    aState3.update(new CSN(looser2T2, 0, myId2));
    aState3.update(new CSN(looser2T3, 0, myId3));
    ServerState aState3 = newServerState(
        new CSN(looser2T1, 0, myId1),
        new CSN(looser2T2, 0, myId2),
        new CSN(looser2T3, 0, myId3));
    if (looser2IsLocal)
      ReplicationServer.onlyForTestsAddlocalReplicationServer(LOOSER2);

@@ -700,7 +608,6 @@
      throws Exception
  {
    String testCase = "test3ServersMoreCriteria";

    debugInfo("Starting " + testCase);

    // definitions for server names
@@ -709,24 +616,20 @@
    final String LOOSER2 = "localhost:789";

    // Create my state
    ServerState mySt = new ServerState();
    mySt.update(new CSN(4, 0, myId1));
    ServerState mySt = newServerState(new CSN(4, 0, myId1));

    // State for server 1
    ServerState aState1 = new ServerState();
    aState1.update(new CSN(looser1T1, 0, myId1));
    ServerState aState1 = newServerState(new CSN(looser1T1, 0, myId1));
    if (looser1IsLocal)
      ReplicationServer.onlyForTestsAddlocalReplicationServer(LOOSER1);

    // State for server 2
    ServerState aState2 = new ServerState();
    aState2.update(new CSN(winnerT1, 0, myId1));
    ServerState aState2 = newServerState(new CSN(winnerT1, 0, myId1));
    if (winnerIsLocal)
      ReplicationServer.onlyForTestsAddlocalReplicationServer(WINNER);

    // State for server 3
    ServerState aState3 = new ServerState();
    aState3.update(new CSN(looser2T1, 0, myId1));
    ServerState aState3 = newServerState(new CSN(looser2T1, 0, myId1));
    if (looser2IsLocal)
      ReplicationServer.onlyForTestsAddlocalReplicationServer(LOOSER2);

@@ -1424,7 +1327,6 @@
      throws Exception
  {
    String testCase = "testComputeBestServerForWeight";

    debugInfo("Starting " + testCase);

    final RSEvaluations evals = new RSEvaluations(localServerId, servers);