In ReplicationBroker, removed null checks on the replication domain. Had to...

Jean-Noel Rouvignac

20.17.2013 e26f8c815fd96beab297c7a58c90d7ae338eaa87

In ReplicationBroker, removed null checks on the replication domain.
Had to change integrated unit tests to pass a dummy replication domain.


ReplicationBroker.java:
Removed null checks on the domain.

DummyReplicationDomain.java: ADDED

ReplicationTestCase.java:
Removed the replicationDomain parameter which were always passed as null.
Could remove one openReplicationSession() method.
Passed in a DummyReplicationDomain when creating a ReplicationBroker.


ExternalChangeLogTest.java, ReplicationServerTest.java, HistoricalCsnOrderingTest.java, StateMachineTest.java:
Passed in a DummyReplicationDomain when creating a ReplicationBroker.
Also consequence of removing the replicationDomain parameter in ReplicationTestCase.openReplicationSession().

 opendj-sdk/opends/src/server/org/opends/server/replication/service/ReplicationBroker.java

@@ -359,11 +359,7 @@
   */
  private long getGenerationID()
  {
    if (domain != null)
    {
      // Update the generation id
      generationID = domain.getGenerationID();
    }
    generationID = domain.getGenerationID();
    return generationID;
  }

@@ -842,15 +838,12 @@
   */
  private void connectAsDataServer()
  {
    if (domain != null)
    {
      /*
      If a first connect or a connection failure occur, we go through here.
      force status machine to NOT_CONNECTED_STATUS so that monitoring can
      see that we are not connected.
      */
      domain.toNotConnectedStatus();
    }
    /*
     * If a first connect or a connection failure occur, we go through here.
     * force status machine to NOT_CONNECTED_STATUS so that monitoring can see
     * that we are not connected.
     */
    domain.toNotConnectedStatus();

    /*
    Stop any existing heartbeat monitor and changeTime publisher
@@ -1019,10 +1012,7 @@
      sendWindow = new Semaphore(maxSendWindow);
      rcvWindow = getMaxRcvWindow();

      if (domain != null)
      {
        domain.sessionInitiated(initStatus, rsInfo.getServerState());
      }
      domain.sessionInitiated(initStatus, rsInfo.getServerState());

      final byte groupId = getGroupId();
      if (rs.getGroupId() != groupId)
@@ -1318,23 +1308,15 @@
    {
      // Send our StartSessionMsg.
      final StartSessionMsg startSessionMsg;
      if (domain != null)
      {
        startSessionMsg = new StartSessionMsg(
      startSessionMsg = new StartSessionMsg(
          initStatus,
          domain.getRefUrls(),
          domain.isAssured(),
          domain.getAssuredMode(),
          domain.getAssuredSdLevel());
        startSessionMsg.setEclIncludes(
            domain.getEclIncludes(domain.getServerId()),
            domain.getEclIncludesForDeletes(domain.getServerId()));
      }
      else
      {
        startSessionMsg =
          new StartSessionMsg(initStatus, new ArrayList<String>());
      }
      startSessionMsg.setEclIncludes(
          domain.getEclIncludes(domain.getServerId()),
          domain.getEclIncludesForDeletes(domain.getServerId()));
      final Session session = electedRS.session;
      session.publish(startSessionMsg);

@@ -3094,13 +3076,10 @@
    // Remove any replication server that may have disappeared from the topology
    replicationServerInfos.keySet().retainAll(rssToKeep);

    if (domain != null)
    for (DSInfo info : dsList)
    {
      for (DSInfo info : dsList)
      {
        domain.setEclIncludes(info.getDsId(), info.getEclIncludes(),
            info.getEclIncludesForDeletes());
      }
      domain.setEclIncludes(info.getDsId(), info.getEclIncludes(),
          info.getEclIncludesForDeletes());
    }
  }


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

@@ -52,7 +52,6 @@
import org.opends.server.replication.server.ReplicationServer;
import org.opends.server.replication.server.changelog.je.JEChangelogDB;
import org.opends.server.replication.service.ReplicationBroker;
import org.opends.server.replication.service.ReplicationDomain;
import org.opends.server.schema.IntegerSyntax;
import org.opends.server.types.*;
import org.testng.annotations.AfterClass;
@@ -196,7 +195,7 @@
      boolean emptyOldChanges) throws Exception
  {
    return openReplicationSession(baseDN, serverId, windowSize,
        port, timeout, emptyOldChanges, getGenerationId(baseDN), null);
        port, timeout, emptyOldChanges, getGenerationId(baseDN));
  }

  /**
@@ -207,36 +206,21 @@
      int serverId, int windowSize, int port, int timeout,
      boolean emptyOldChanges, long generationId) throws Exception
  {
    return openReplicationSession(baseDN, serverId, windowSize,
        port, timeout, emptyOldChanges, generationId, null);
  }

  /**
   * Open a replicationServer session to the local ReplicationServer
   * providing the generationId.
   */
  protected ReplicationBroker openReplicationSession(final DN baseDN,
      int serverId, int windowSize, int port, int timeout,
      boolean emptyOldChanges, long generationId,
      ReplicationDomain replicationDomain) throws Exception
  {
    DomainFakeCfg config = newFakeCfg(baseDN, serverId, port);
    config.setWindowSize(windowSize);
    return openReplicationSession(config, port, timeout, emptyOldChanges,
        generationId, replicationDomain);
    return openReplicationSession(config, port, timeout, emptyOldChanges, generationId);
  }

  protected ReplicationBroker openReplicationSession(ReplicationDomainCfg config,
      int port, int timeout, boolean emptyOldChanges, long generationId,
      ReplicationDomain replicationDomain) throws Exception
      int port, int timeout, boolean emptyOldChanges, long generationId) throws Exception
  {
    ServerState state = new ServerState();

    if (emptyOldChanges)
      new PersistentServerState(config.getBaseDN(), config.getServerId(), new ServerState());

    ReplicationBroker broker = new ReplicationBroker(replicationDomain, state,
        config, generationId, getReplSessionSecurity());
    final ReplicationBroker broker = new ReplicationBroker(
        new DummyReplicationDomain(generationId), state, config, generationId, getReplSessionSecurity());
    connect(broker, port, timeout);
    return broker;
  }

 opendj-sdk/opends/tests/unit-tests-testng/src/server/org/opends/server/replication/plugin/DummyReplicationDomain.java

New file
@@ -0,0 +1,102 @@
/*
 * 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 legal-notices/CDDLv1_0.txt
 * or http://forgerock.org/license/CDDLv1.0.html.
 * 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 legal-notices/CDDLv1_0.txt.
 * 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 2013 ForgeRock AS
 */
package org.opends.server.replication.plugin;

import java.io.InputStream;
import java.io.OutputStream;
import java.util.Set;
import java.util.concurrent.atomic.AtomicBoolean;

import org.opends.server.replication.common.ServerState;
import org.opends.server.replication.common.ServerStatus;
import org.opends.server.replication.common.StatusMachineEvent;
import org.opends.server.replication.protocol.UpdateMsg;
import org.opends.server.replication.service.ReplicationDomain;
import org.opends.server.types.DirectoryException;

@SuppressWarnings("javadoc")
public class DummyReplicationDomain extends ReplicationDomain
{

  private final long generationId;

  public DummyReplicationDomain(long generationId)
  {
    super(null, -1, 0);
    this.generationId = generationId;
  }

  @Override
  public String getBaseDNString()
  {
    return "DummyReplicationDomain";
  }

  @Override
  protected void setNewStatus(StatusMachineEvent event)
  {
  }

  @Override
  public void sessionInitiated(ServerStatus initStatus, ServerState rsState)
  {
  }

  @Override
  public boolean setEclIncludes(int serverId, Set<String> includeAttributes,
      Set<String> includeAttributesForDeletes)
  {
    return false;
  }

  @Override
  protected void exportBackend(OutputStream output) throws DirectoryException
  {
  }

  @Override
  protected void importBackend(InputStream input) throws DirectoryException
  {
  }

  @Override
  public long countEntries() throws DirectoryException
  {
    return 0;
  }

  @Override
  public boolean processUpdate(UpdateMsg updateMsg, AtomicBoolean shutdown)
  {
    return false;
  }

  @Override
  public long getGenerationID()
  {
    return this.generationId;
  }

}

 opendj-sdk/opends/tests/unit-tests-testng/src/server/org/opends/server/replication/plugin/HistoricalCsnOrderingTest.java

@@ -70,7 +70,8 @@

    public TestBroker(List<ReplicationMsg> list)
    {
      super(null, null, new DomainFakeCfg(null, 0, TestCaseUtils.<String> newSortedSet()), 0, null);
      super(new DummyReplicationDomain(0), null,
          new DomainFakeCfg(null, 0, TestCaseUtils.<String> newSortedSet()), 0, null);
      this.list = list;
    }


 opendj-sdk/opends/tests/unit-tests-testng/src/server/org/opends/server/replication/plugin/StateMachineTest.java

@@ -28,7 +28,10 @@

import java.io.IOException;
import java.net.SocketTimeoutException;
import java.util.*;
import java.util.ArrayList;
import java.util.List;
import java.util.SortedSet;
import java.util.TreeSet;
import java.util.concurrent.atomic.AtomicBoolean;

import org.opends.messages.Category;
@@ -228,7 +231,8 @@
    fakeCfg.setHeartbeatInterval(0);
    fakeCfg.setChangetimeHeartbeatInterval(500);
    ReplSessionSecurity security = new ReplSessionSecurity(null, null, null, true);
    ReplicationBroker broker = new ReplicationBroker(null, state, fakeCfg, generationId, security);
    ReplicationBroker broker = new ReplicationBroker(
        new DummyReplicationDomain(generationId), state, fakeCfg, generationId, security);
    broker.start();
    checkConnection(30, broker, rs1Port);
    return broker;

 opendj-sdk/opends/tests/unit-tests-testng/src/server/org/opends/server/replication/server/ExternalChangeLogTest.java

@@ -1540,13 +1540,13 @@
      DomainFakeCfg config1 = newFakeCfg(TEST_ROOT_DN, SERVER_ID_1, replicationServerPort);
      config1.setChangetimeHeartbeatInterval(100); // ms
      server01 = openReplicationSession(config1, replicationServerPort,
          brokerSessionTimeout, true, getGenerationId(TEST_ROOT_DN), null);
              brokerSessionTimeout, true, getGenerationId(TEST_ROOT_DN));

      // Create broker on o=test2
      DomainFakeCfg config2 = newFakeCfg(TEST_ROOT_DN2, SERVER_ID_2, replicationServerPort);
      config2.setChangetimeHeartbeatInterval(100); //ms
      server02 = openReplicationSession(config2, replicationServerPort,
          brokerSessionTimeout, true, EMPTY_DN_GENID, null);
              brokerSessionTimeout, true, EMPTY_DN_GENID);

      int ts = 1;
      // Produce update 1

 opendj-sdk/opends/tests/unit-tests-testng/src/server/org/opends/server/replication/server/ReplicationServerTest.java

@@ -46,6 +46,7 @@
import org.opends.server.replication.common.CSNGenerator;
import org.opends.server.replication.common.ServerState;
import org.opends.server.replication.common.ServerStatus;
import org.opends.server.replication.plugin.DummyReplicationDomain;
import org.opends.server.replication.plugin.MultimasterReplication;
import org.opends.server.replication.plugin.ReplicationServerListener;
import org.opends.server.replication.protocol.*;
@@ -355,8 +356,10 @@

    // Connect to the replicationServer using the state created above.
    try {
      broker = new ReplicationBroker(null, state, newFakeCfg(TEST_ROOT_DN, 3, replicationServerPort),
          getGenerationId(TEST_ROOT_DN), getReplSessionSecurity());
      final long generationId = getGenerationId(TEST_ROOT_DN);
      broker = new ReplicationBroker(new DummyReplicationDomain(generationId),
          state, newFakeCfg(TEST_ROOT_DN, 3, replicationServerPort),
          generationId, getReplSessionSecurity());
      connect(broker, replicationServerPort, 5000);

      ReplicationMsg receivedMsg = broker.receive();