Fix OPENDJ-1354 - replication threads BLOCKED in pendingChanges queue

Matthew Swift

25.02.2014 21af6610b07617ecbf1b826310a2f244deb4d348

 opendj3-server-dev/src/server/org/opends/server/replication/server/ServerHandler.java

@@ -27,7 +27,9 @@
package org.opends.server.replication.server;

import java.io.IOException;

import org.forgerock.i18n.slf4j.LocalizedLogger;

import java.util.List;
import java.util.Random;
import java.util.concurrent.Semaphore;
@@ -45,6 +47,7 @@
import org.opends.server.replication.protocol.*;
import org.opends.server.types.*;
import org.forgerock.opendj.ldap.ResultCode;

import static org.opends.messages.ReplicationMessages.*;

/**
@@ -485,7 +488,7 @@
   */
  public ReplicationServerDomain getDomain()
  {
    return this.replicationServerDomain;
    return replicationServerDomain;
  }

  /**
@@ -848,21 +851,45 @@
   *
   * @param msg The message to be processed.
   */
  public void process(RoutableMsg msg)
  void process(RoutableMsg msg)
  {
    if (logger.isTraceEnabled())
    {
      logger.trace("In "
          + replicationServerDomain.getLocalRSMonitorInstanceName() + " "
          + this + " processes routable msg received:" + msg);
    }
    replicationServerDomain.process(msg, this);
  }

  /**
   * Responds to a monitor request message.
   *
   * @param msg
   *          The monitor request message.
   */
  void processMonitorRequestMsg(MonitorRequestMsg msg)
  {
    replicationServerDomain.processMonitorRequestMsg(msg, this);
  }

  /**
   * Responds to a monitor message.
   *
   * @param msg
   *          The monitor message.
   */
  void processMonitorMsg(MonitorMsg msg)
  {
    replicationServerDomain.processMonitorMsg(msg, this);
  }

  /**
   * Processes a change time heartbeat msg.
   *
   * @param msg The message to be processed.
   */
  public void process(ChangeTimeHeartbeatMsg msg)
  void process(ChangeTimeHeartbeatMsg msg)
  {
    if (logger.isTraceEnabled())
      logger.trace("In "
@@ -926,15 +953,6 @@
  }

  /**
   * Sets the replication server domain associated.
   * @param rsd The provided replication server domain.
   */
  protected void setReplicationServerDomain(ReplicationServerDomain rsd)
  {
    this.replicationServerDomain = rsd;
  }

  /**
   * Sets the window size when used when sending to the remote.
   * @param size The provided window size.
   */
@@ -1180,7 +1198,7 @@
   * Process a Ack message received.
   * @param ack the message received.
   */
  public void processAck(AckMsg ack)
  void processAck(AckMsg ack)
  {
    if (replicationServerDomain!=null)
      replicationServerDomain.processAck(ack, this);
@@ -1201,7 +1219,7 @@
   * Process a ResetGenerationIdMsg message received.
   * @param msg the message received.
   */
  public void processResetGenId(ResetGenerationIdMsg msg)
  void processResetGenId(ResetGenerationIdMsg msg)
  {
    if (replicationServerDomain!=null)
      replicationServerDomain.resetGenerationId(this, msg);