Fix OPENDJ-115: Make replication connection timeouts and various monitoring...

Matthew Swift

07.09.2011 ce949f4b26b4c2adcaae36784180e15a3d68abde

Fix OPENDJ-115: Make replication connection timeouts and various monitoring intervals configurable

Added connection-timeout property to replication synchronization provider configuration entry which is used for connection and SSL negotiation. Previously the timeout was hard-coded.

 opends/resource/schema/02-config.ldif

@@ -3609,7 +3609,8 @@
  NAME 'ds-cfg-replication-synchronization-provider'
  SUP ds-cfg-synchronization-provider
  STRUCTURAL
  MAY ( ds-cfg-num-update-replay-threads )
  MAY ( ds-cfg-num-update-replay-threads $
        ds-cfg-connection-timeout )
  X-ORIGIN 'OpenDS Directory Server' )
objectClasses: ( 1.3.6.1.4.1.26027.1.2.94
  NAME 'ds-cfg-dictionary-password-validator'

 opends/src/admin/defn/org/opends/server/admin/std/ReplicationSynchronizationProviderConfiguration.xml

@@ -104,4 +104,24 @@
      </ldap:attribute>
    </adm:profile>
  </adm:property>

  <adm:property name="connection-timeout" advanced="true">
    <adm:synopsis>
      Specifies the timeout used when connecting to peers and when performing
      SSL negotiation.
    </adm:synopsis>
    <adm:default-behavior>
      <adm:defined>
        <adm:value>5 seconds</adm:value>
      </adm:defined>
    </adm:default-behavior>
    <adm:syntax>
      <adm:duration base-unit="ms" lower-limit="0" />
    </adm:syntax>
    <adm:profile name="ldap">
      <ldap:attribute>
        <ldap:name>ds-cfg-connection-timeout</ldap:name>
      </ldap:attribute>
    </adm:profile>
  </adm:property>
</adm:managed-object>

 opends/src/admin/messages/ReplicationSynchronizationProviderCfgDefn.properties

@@ -1,6 +1,7 @@
user-friendly-name=Replication Synchronization Provider
user-friendly-plural-name=Replication Synchronization Providers
synopsis=The Replication Synchronization Provider provides multi-master replication of data across multiple directory server instances.
property.connection-timeout.synopsis=Specifies the timeout used when connecting to peers and when performing SSL negotiation.
property.enabled.synopsis=Indicates whether the Replication Synchronization Provider is enabled for use.
property.java-class.synopsis=Specifies the fully-qualified name of the Java class that provides the Replication Synchronization Provider implementation.
property.num-update-replay-threads.synopsis=Specifies the number of update replay threads.

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

@@ -123,6 +123,11 @@
  private static boolean initializationCompleted = true;

  /**
   * The configurable connection/handshake timeout.
   */
  private static volatile int connectionTimeoutMS = 5000;

  /**
   * Finds the domain for a given DN.
   *
   * @param dn         The DN for which the domain must be returned.
@@ -285,6 +290,8 @@
    configuration.addReplicationChangeListener(this);

    replayThreadNumber = configuration.getNumUpdateReplayThreads();
    connectionTimeoutMS = (int) Math.min(configuration.getConnectionTimeout(),
        Integer.MAX_VALUE);

    //  Create the list of domains that are already defined.
    for (String name : configuration.listReplicationDomains())
@@ -821,6 +828,9 @@
      createReplayThreads();
    }

    connectionTimeoutMS = (int) Math.min(configuration.getConnectionTimeout(),
        Integer.MAX_VALUE);

    return new ConfigChangeResult(ResultCode.SUCCESS, false);
  }

@@ -882,4 +892,14 @@
    return false;
  }

  /**
   * Returns the connection timeout in milli-seconds.
   *
   * @return The connection timeout in milli-seconds.
   */
  public static int getConnectionTimeoutMS()
  {
    return connectionTimeoutMS;
  }

}

 opends/src/server/org/opends/server/replication/protocol/ReplSessionSecurity.java

@@ -77,17 +77,6 @@
   */
  private final String sslCipherSuites[];

  /**
   * The default soTimeout value to be used at handshake phases. (DS<->RS and
   * RS<->RS)
   */
  public static final int HANDSHAKE_TIMEOUT = 4000;

  /**
   * The default connection timeout.
   */
  public static final int CONNECTION_TIMEOUT = 4000;



  /**

 opends/src/server/org/opends/server/replication/server/ReplicationServer.java

@@ -60,6 +60,7 @@
import org.opends.server.core.networkgroups.NetworkGroup;
import org.opends.server.loggers.debug.DebugTracer;
import org.opends.server.replication.common.*;
import org.opends.server.replication.plugin.MultimasterReplication;
import org.opends.server.replication.protocol.*;
import org.opends.server.types.*;
import org.opends.server.util.LDIFReader;
@@ -307,9 +308,9 @@
          newSocket = listenSocket.accept();
          newSocket.setTcpNoDelay(true);
          newSocket.setKeepAlive(true);
          session =
            replSessionSecurity.createServerSession(newSocket,
                ReplSessionSecurity.HANDSHAKE_TIMEOUT);
          int timeoutMS = MultimasterReplication.getConnectionTimeoutMS();
          session = replSessionSecurity.createServerSession(newSocket,
              timeoutMS);
          if (session == null) // Error, go back to accept
            continue;
        }
@@ -490,10 +491,9 @@
      InetSocketAddress ServerAddr = new InetSocketAddress(
          InetAddress.getByName(hostname), Integer.parseInt(port));
      socket.setTcpNoDelay(true);
      socket.connect(ServerAddr, ReplSessionSecurity.CONNECTION_TIMEOUT);

      session = replSessionSecurity.createClientSession(socket,
          ReplSessionSecurity.HANDSHAKE_TIMEOUT);
      int timeoutMS = MultimasterReplication.getConnectionTimeoutMS();
      socket.connect(ServerAddr, timeoutMS);
      session = replSessionSecurity.createClientSession(socket, timeoutMS);

      ReplicationServerHandler handler = new ReplicationServerHandler(
          session, queueSize, this.serverURL, serverId, this,

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

@@ -68,6 +68,7 @@
import org.opends.server.replication.common.RSInfo;
import org.opends.server.replication.common.ServerState;
import org.opends.server.replication.common.ServerStatus;
import org.opends.server.replication.plugin.MultimasterReplication;
import org.opends.server.replication.protocol.*;
import org.opends.server.types.DebugLogLevel;
import org.opends.server.util.ServerConstants;
@@ -1198,9 +1199,9 @@
      socket = new Socket();
      socket.setReceiveBufferSize(1000000);
      socket.setTcpNoDelay(true);
      socket.connect(serverAddr, ReplSessionSecurity.CONNECTION_TIMEOUT);
      localSession = replSessionSecurity.createClientSession(
          socket, ReplSessionSecurity.HANDSHAKE_TIMEOUT);
      int timeoutMS = MultimasterReplication.getConnectionTimeoutMS();
      socket.connect(serverAddr, timeoutMS);
      localSession = replSessionSecurity.createClientSession(socket, timeoutMS);
      boolean isSslEncryption = replSessionSecurity
          .isSslEncryption(server);


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

@@ -397,8 +397,9 @@
          newSocket.setKeepAlive(true);
          // Create client session
          ReplSessionSecurity replSessionSecurity = new ReplSessionSecurity();
          int timeoutMS = MultimasterReplication.getConnectionTimeoutMS();
          session = replSessionSecurity.createServerSession(newSocket,
            ReplSessionSecurity.HANDSHAKE_TIMEOUT);
              timeoutMS);
          if (session == null) // Error, go back to accept
          {
            continue;

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

@@ -23,6 +23,7 @@
 *
 *
 *      Copyright 2009 Sun Microsystems, Inc.
 *      Portions copyright 2011 ForgeRock AS.
 */
package org.opends.server.replication.plugin;

@@ -182,4 +183,9 @@
    return null;
  }

  public long getConnectionTimeout()
  {
    return 5000;
  }

}

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

@@ -54,6 +54,7 @@
import org.opends.server.replication.common.AssuredMode;
import org.opends.server.replication.common.ServerState;
import org.opends.server.replication.common.ServerStatus;
import org.opends.server.replication.plugin.MultimasterReplication;
import org.opends.server.replication.protocol.ProtocolSession;
import org.opends.server.replication.protocol.ProtocolVersion;
import org.opends.server.replication.protocol.ReplServerStartMsg;
@@ -953,14 +954,14 @@
          new InetSocketAddress("localhost", port);
        Socket socket = new Socket();
        socket.setTcpNoDelay(true);
        socket.connect(serverAddr, ReplSessionSecurity.CONNECTION_TIMEOUT);
        int timeoutMS = MultimasterReplication.getConnectionTimeoutMS();
        socket.connect(serverAddr, timeoutMS);

        // Create client session
        fakePort++;
        String fakeUrl = "localhost:" + fakePort;
        ReplSessionSecurity replSessionSecurity = new ReplSessionSecurity();
        session = replSessionSecurity.createClientSession(socket,
          ReplSessionSecurity.HANDSHAKE_TIMEOUT);
        session = replSessionSecurity.createClientSession(socket, timeoutMS);

        // Send our repl server start msg
        ReplServerStartMsg replServerStartMsg = new ReplServerStartMsg(serverId,

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

@@ -1013,11 +1013,11 @@
    Socket socket = new Socket();
    socket.setReceiveBufferSize(1000000);
    socket.setTcpNoDelay(true);
    socket.connect(ServerAddr, ReplSessionSecurity.CONNECTION_TIMEOUT);
    int timeoutMS = MultimasterReplication.getConnectionTimeoutMS();
    socket.connect(ServerAddr, timeoutMS);
    ReplSessionSecurity replSessionSecurity = getReplSessionSecurity();
    ProtocolSession session =
         replSessionSecurity.createClientSession(socket,
         ReplSessionSecurity.HANDSHAKE_TIMEOUT);
    ProtocolSession session = replSessionSecurity.createClientSession(socket,
        timeoutMS);

    boolean sslEncryption =
         DirectoryConfig.getCryptoManager().isSslEncryption();