DistributedVirtualDatabase.java

/**
 * C-JDBC: Clustered JDBC.
 * Copyright (C) 2002-2005 French National Institute For Research In Computer
 * Science And Control (INRIA).
 * Contact: c-jdbc@objectweb.org
 * 
 * This library is free software; you can redistribute it and/or modify it
 * under the terms of the GNU Lesser General Public License as published by the
 * Free Software Foundation; either version 2.1 of the License, or any later
 * version.
 * 
 * This library is distributed in the hope that it will be useful, but WITHOUT
 * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
 * FITNESS FOR A PARTICULAR PURPOSE. See the GNU Lesser General Public License
 * for more details.
 * 
 * You should have received a copy of the GNU Lesser General Public License
 * along with this library; if not, write to the Free Software Foundation,
 * Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA.
 *
 * Initial developer(s): Emmanuel Cecchet. 
 * Contributor(s): _________________________.
 */

package org.objectweb.cjdbc.controller.virtualdatabase;

import java.io.Serializable;
import java.net.URL;
import java.sql.SQLException;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.Hashtable;
import java.util.Iterator;
import java.util.Map.Entry;

import javax.management.NotCompliantMBeanException;

import org.dom4j.DocumentException;
import org.objectweb.cjdbc.common.exceptions.ControllerException;
import org.objectweb.cjdbc.common.exceptions.VirtualDatabaseException;
import org.objectweb.cjdbc.common.i18n.Translate;
import org.objectweb.cjdbc.common.jmx.JmxException;
import org.objectweb.cjdbc.common.jmx.notifications.CjdbcNotificationList;
import org.objectweb.cjdbc.common.shared.BackendInfo;
import org.objectweb.cjdbc.common.sql.filters.AbstractBlobFilter;
import org.objectweb.cjdbc.common.xml.DatabasesXmlTags;
import org.objectweb.cjdbc.controller.authentication.AuthenticationManager;
import org.objectweb.cjdbc.controller.backend.BackendRecoveryPolicy;
import org.objectweb.cjdbc.controller.backend.DatabaseBackend;
import org.objectweb.cjdbc.controller.core.Controller;
import org.objectweb.cjdbc.controller.jmx.MBeanServerManager;
import org.objectweb.cjdbc.controller.jmx.RmiConnector;
import org.objectweb.cjdbc.controller.requestmanager.RequestManager;
import org.objectweb.cjdbc.controller.requestmanager.distributed.DistributedRequestManager;
import org.objectweb.cjdbc.controller.virtualdatabase.protocol.BackendStatus;
import org.objectweb.cjdbc.controller.virtualdatabase.protocol.BackendTransfer;
import org.objectweb.cjdbc.controller.virtualdatabase.protocol.CJDBCGroupMessage;
import org.objectweb.cjdbc.controller.virtualdatabase.protocol.Commit;
import org.objectweb.cjdbc.controller.virtualdatabase.protocol.ControllerName;
import org.objectweb.cjdbc.controller.virtualdatabase.protocol.DistributedRequest;
import org.objectweb.cjdbc.controller.virtualdatabase.protocol.EnableBackend;
import org.objectweb.cjdbc.controller.virtualdatabase.protocol.Rollback;
import org.objectweb.cjdbc.controller.virtualdatabase.protocol.VirtualDatabaseConfiguration;
import org.objectweb.tribe.adapters.MulticastRequestAdapter;
import org.objectweb.tribe.adapters.MulticastRequestListener;
import org.objectweb.tribe.adapters.MulticastResponse;
import org.objectweb.tribe.channel.JGroupsReliableChannelWithGms;
import org.objectweb.tribe.channel.ReliableGroupChannelWithGms;
import org.objectweb.tribe.common.Address;
import org.objectweb.tribe.common.Group;
import org.objectweb.tribe.common.GroupIdentifier;
import org.objectweb.tribe.common.Member;
import org.objectweb.tribe.exceptions.ChannelException;
import org.objectweb.tribe.exceptions.NotConnectedException;
import org.objectweb.tribe.exceptions.TimeoutException;
import org.objectweb.tribe.gms.JGroupsMembershipService;
import org.objectweb.tribe.messages.MessageListener;

/**
 * A <code>DistributedVirtualDatabase</code> is a virtual database hosted by
 * several controllers. Communication between the controllers is achieved with
 * reliable multicast provided by Javagroups.
 * 
 * @author <a href="mailto:Emmanuel.Cecchet@inria.fr">Emmanuel Cecchet </a>
 * @version 1.0
 */
public class DistributedVirtualDatabase extends VirtualDatabase
    implements
      MessageListener,
      MulticastRequestListener
{
  //
  // How the code is organized ?
  //
  // 1. Member variables
  // 2. Constructor(s)
  // 3. Request handling
  // 4. Transaction handling
  // 5. Database backend management
  // 6. Distribution management (multicast)
  // 7. Getter/Setter (possibly in alphabetical order)
  //

  // Distribution

  /** Group name */
  private String                      groupName      = null;
  /** Controller name mapping: jgroup Addr --> controller Name */
  private Hashtable                   controllersMap;
  /** backendName -><code>BackendRecoveryPolicy</code> */
  private HashMap                     backendRecoveryPolicy;
  /** JGroups Address -><code>ArrayList</code> of <code>DatabaseBackend</code> */
  private Hashtable                   backendsPerController;

  /** JGroups channel */
  private ReliableGroupChannelWithGms channel        = null;
  /** MessageDispatcher to communicate with the group */
  private MulticastRequestAdapter     multicastRequestAdapter;
  private Group                       currentGroup   = null;
  private ArrayList                   allMemberButUs = null;

  //
  // Constructors
  //

  /**
   * Creates a new <code>DistributedVirtualDatabase</code> instance.
   * 
   * @param controller the controller we belong to
   * @param name the virtual database name
   * @param groupName the virtual database group name
   * @param maxConnections maximum number of concurrent connections.
   * @param pool should we use a pool of threads for handling connections?
   * @param minThreads minimum number of threads in the pool
   * @param maxThreads maximum number of threads in the pool
   * @param maxThreadIdleTime maximum time a thread can remain idle before being
   *          removed from the pool.
   * @param sqlShortFormLength maximum number of characters of an SQL statement
   *          to diplay in traces or exceptions
   * @param blobFilter encoding method for blobs
   * @exception NotCompliantMBeanException in case the bean does not comply with
   *              jmx
   * @exception JmxException the bean could not be registeed
   */
  public DistributedVirtualDatabase(Controller controller, String name,
      String groupName, int maxConnections, boolean pool, int minThreads,
      int maxThreads, long maxThreadIdleTime, int sqlShortFormLength,
      AbstractBlobFilter blobFilter) throws NotCompliantMBeanException,
      JmxException
  {
    super(controller, name, maxConnections, pool, minThreads, maxThreads,
        maxThreadIdleTime, sqlShortFormLength, blobFilter);

    this.groupName = groupName;
    backendRecoveryPolicy = new HashMap();
    backendsPerController = new Hashtable();
    controllersMap = new Hashtable();
  }

  /**
   * Disconnect the channel and close it.
   * 
   * @see java.lang.Object#finalize()
   */
  protected void finalize() throws Throwable
  {
    quitChannel();
    super.finalize();
  }

  /**
   * @see org.objectweb.cjdbc.controller.virtualdatabase.VirtualDatabase#addBackend(org.objectweb.cjdbc.controller.backend.DatabaseBackend)
   */
  public void addBackend(DatabaseBackend db) throws VirtualDatabaseException
  {
    // Add the backend to the virtual database.
    super.addBackend(db);

    // Send a group message if already joined group
    if (allMemberButUs != null)
    {
      try
      {
        broadcastBackendInformation(allMemberButUs);
      }
      catch (Exception e)
      {
        String msg = "Error while broadcasting backend information when adding backend";
        logger.error(msg, e);
        throw new VirtualDatabaseException(msg, e);
      }
    }
  }

  /**
   * Quit the jgroups channel
   * 
   * @throws NotConnectedException if the channel is not connected
   * @throws ChannelException if an error occured while closing the channel
   */
  public void quitChannel() throws ChannelException, NotConnectedException
  {
    if (channel != null)
    {
      channel.close();
    }
  }

  /**
   * Returns the controllerName value.
   * 
   * @return Returns the controllerName.
   */
  public String getControllerName()
  {
    return controller.getControllerName();
  }

  /**
   * Returns the group name this virtual database belongs to.
   * 
   * @return a <code>String</code> value. Returns <code>null</code> if this
   *         virtual database is standalone
   */
  public String getGroupName()
  {
    return groupName;
  }

  /**
   * Sets the group name used by the controllers hosting this virtual database.
   * 
   * @param groupName the group name to set
   */
  public void setGroupName(String groupName)
  {
    this.groupName = groupName;
  }

  /**
   * Add a BackendRecoveryPolicy
   * 
   * @param policy the policy to add
   */
  public void addBackendRecoveryPolicy(BackendRecoveryPolicy policy)
  {
    backendRecoveryPolicy.put(policy.getBackendName(), policy);
  }

  /**
   * Sets a new distributed request manager for this database.
   * 
   * @param requestManager the new request manager.
   */
  public void setRequestManager(RequestManager requestManager)
  {
    if (!(requestManager instanceof DistributedRequestManager))
      throw new RuntimeException(
          "A distributed virtual database can only work with a distributed request manager.");
    this.requestManager = requestManager;
  }

  //
  //  Distribution handling
  //

  /**
   * Makes this virtual database join a virtual database group. Those groups are
   * mapped to JavaGroups groups.
   * 
   * @exception Exception if an error occurs
   */
  public void joinGroup() throws Exception
  {
    try
    {
      // Read the protocol stack configuration from jgroups.xml
      URL jgroupConfigFile = DistributedVirtualDatabase.class
          .getResource("/jgroups.xml");
      if (jgroupConfigFile == null)
        logger.warn(Translate
            .get("virtualdatabase.distributed.jgroups.xml.not.found"));
      else
        logger.info(Translate.get("virtualdatabase.distributed.jgroups.using",
            jgroupConfigFile.toString()));
      JGroupsMembershipService gms = new JGroupsMembershipService(
          jgroupConfigFile);
      channel = new JGroupsReliableChannelWithGms(gms);
      if (logger.isDebugEnabled())
        logger.debug("Group communication channel is configured as follows: "
            + ((JGroupsReliableChannelWithGms) channel).getProperties());

      // Join the group
      channel.join(new Group(new GroupIdentifier(groupName)));
      multicastRequestAdapter = new MulticastRequestAdapter(channel // group
          // channel
          , this /* MessageListener */
          , this /* MulticastRequestListener */
      );

      // Let the MulticastRequestAdapter thread pump the membership out of the
      // JGroups channel.
      Thread.yield();

      logger.info("Group " + groupName + " connected to "
          + channel.getLocalMembership());

      // Add ourselves to the list of controllers
      controllersMap.put(channel.getLocalMembership(), controller.getJmxName());

      // Check if we are alone or not
      currentGroup = channel.getCurrentGroup();
      if (currentGroup.getMembers().size() == 1)
      {
        logger.info(Translate.get(
            "virtualdatabase.distributed.configuration.first.in.group",
            groupName));
        allMemberButUs = new ArrayList();
        return;
      }

      // Compute the group list without us
      allMemberButUs = (ArrayList) currentGroup.getMembers().clone();

      // Our controller id is our position in the membership.
      // This assumes that all controllers have the view in the same order.
      ((DistributedRequestManager) requestManager)
          .setControllerId(allMemberButUs.indexOf(channel.getLocalMembership()));
      allMemberButUs.remove(channel.getLocalMembership());

      // Check with the other controller that our config is compatible
      if (checkConfigurationCompatibility(allMemberButUs) == false)
      {
        String msg = Translate
            .get("virtualdatabase.distributed.configuration.not.compatible");
        logger.error(msg);
        throw new ControllerException(msg);
      }

      logger.info(Translate
          .get("virtualdatabase.distributed.configuration.compatible"));

      // Distribute backends among controllers knowing that at this point
      // there is no conflict on the backend distribution policies.
      broadcastBackendInformation(allMemberButUs);
    }
    catch (Exception e)
    {
      String msg = Translate.get("virtualdatabase.distributed.joingroup.error",
          groupName);
      if (e instanceof RuntimeException)
        logger.error(msg, e);
      throw new Exception(msg + " (" + e + ")");
    }
  }

  /**
   * Get the JGroups' channel used for group communications
   * 
   * @return a <code>JChannel</code>
   */
  public ReliableGroupChannelWithGms getChannel()
  {
    return channel;
  }

  /**
   * Return the group communication multicast request adapter.
   * 
   * @return the group communication multicast request adapter
   */
  public MulticastRequestAdapter getMulticastRequestAdapter()
  {
    return multicastRequestAdapter;
  }

  /**
   * Returns the currentGroup value.
   * 
   * @return Returns the currentGroup.
   */
  public Group getCurrentGroup()
  {
    return currentGroup;
  }

  /**
   * Returns the list of all members in the group except us.
   * 
   * @return Returns the allMemberButUs.
   */
  public ArrayList getAllMemberButUs()
  {
    return allMemberButUs;
  }

  /**
   * Send the configuration of this controller to remote controller. All remote
   * controllers must agree on the compatibility of the local controller
   * configuration with their own configuration. Compatibility checking include
   * Authentication Manager, Scheduler and Load Balancer settings.
   * 
   * @param dest List of <code>Address</code> to send the message to
   * @return true if the local config is compatible with the group, false
   *         otherwise.
   */
  private boolean checkConfigurationCompatibility(ArrayList dest)
  {
    if (logger.isDebugEnabled())
      logger.debug(Translate
          .get("virtualdatabase.distributed.configuration.checking"));

    // Send our configuration
    MulticastResponse rspList;
    try
    {
      rspList = multicastRequestAdapter.multicastMessage(dest,
          new VirtualDatabaseConfiguration(this),
          MulticastRequestAdapter.WAIT_ALL,
          CJDBCGroupMessage.defaultCastTimeOut);
    }
    catch (TimeoutException e)
    {
      logger.error(
          "Timeout occured while checking configuration compatibility", e);
      return false;
    }
    catch (ChannelException e)
    {
      logger
          .error(
              "Communication error occured while checking configuration compatibility",
              e);
      return false;
    }
    catch (NotConnectedException e)
    {
      logger.error(
          "Channel unavailable while checking configuration compatibility", e);
      return false;
    }

    // Check that everybody agreed
    HashMap results = rspList.getResults();
    int size = results.size();
    if (size == 0)
      logger.warn(Translate
          .get("virtualdatabase.distributed.configuration.checking.noanswer"));
    for (Iterator iter = results.values().iterator(); iter.hasNext();)
    {
      Object response = iter.next();
      if (response instanceof Boolean)
      {
        if (((Boolean) response).booleanValue() == false)
          return false;
      }
      else
      {
        logger
            .error("Unexpected response while checking configuration compatibility: "
                + response);
        return false;
      }
    }

    // Ok, everybody replied true (config is compatible)
    return true;
  }

  /**
   * Broadcast backend information among controllers.
   * 
   * @param dest List of <code>Address</code> to send the message to
   * @throws NotConnectedException if the channel is not connected
   * @throws ChannelException if the channel reported an error
   * @throws TimeoutException if a timeout occured
   */
  private void broadcastBackendInformation(ArrayList dest)
      throws TimeoutException, ChannelException, NotConnectedException
  {
    logger
        .debug(Translate
            .get("virtualdatabase.distributed.configuration.querying.remote.status"));

    // Send our backend status
    MulticastResponse rspList;
    rspList = multicastRequestAdapter.multicastMessage(dest, new BackendStatus(
        getBackendsInfo(backends)), MulticastRequestAdapter.WAIT_ALL,
        CJDBCGroupMessage.defaultCastTimeOut);

    int size = dest.size();
    for (int i = 0; i < size; i++)
    {
      // Add the backend configuration of every remote controller
      Member m = (Member) dest.get(i);
      if (rspList.getResult(m) != null)
      {
        if (logger.isDebugEnabled())
          logger
              .debug(Translate
                  .get(
                      "virtualdatabase.distributed.configuration.updating.backend.list",
                      m.toString()));
      }
      else
        logger.warn(Translate.get(
            "virtualdatabase.distributed.unable.get.remote.status", m
                .toString()));
    }
  }

  /**
   * Check if the given backend definition is compatible with the backend
   * definitions of this distributed virtual database. Not that if the given
   * backend does not exist in the current configuration, it is considered as
   * compatible. Incompatibility results from 2 backends with the same JDBC URL.
   * 
   * @param backend the backend to check
   * @return true if the backend is compatible with the local definition
   * @throws VirtualDatabaseException if locking the local backend list fails
   */
  public boolean isCompatibleBackend(BackendInfo backend)
      throws VirtualDatabaseException
  {
    try
    {
      acquireReadLockBackendLists();
    }
    catch (InterruptedException e)
    {
      String msg = "Unable to acquire read lock on backend list in isCompatibleBackend ("
          + e + ")";
      logger.error(msg);
      throw new VirtualDatabaseException(msg);
    }

    try
    {
      // Find the backend
      String backendURL = backend.getUrl();
      int size = backends.size();
      DatabaseBackend b = null;
      for (int i = 0; i < size; i++)
      {
        b = (DatabaseBackend) backends.get(i);
        if (b.getURL().equals(backendURL))
          return false;
      }
    }
    catch (RuntimeException re)
    {
      throw new VirtualDatabaseException(re);
    }
    finally
    {
      releaseReadLockBackendLists();
    }
    // This backend does not exist here
    return true;
  }

  //
  // Message dispatcher request handling
  //

  /**
   * @see org.objectweb.tribe.messages.MessageListener#receive(java.io.Serializable)
   */
  public void receive(Serializable msg)
  {
    logger.error("Distributed virtual database received unhandled message: "
        + msg);
  }

  /**
   * This method handle the scheduling part of the queries to be sure that the
   * query is scheduled in total order before letting other queries to execute.
   * 
   * @see org.objectweb.tribe.adapters.MulticastRequestListener#handleMessageSingleThreaded(java.io.Serializable,
   *      org.objectweb.tribe.common.Member)
   */
  public Object handleMessageSingleThreaded(Serializable msg, Member sender)
  {
    try
    {
      if (msg != null)
      {
        logger.debug("handleMessageSingleThreaded (" + msg.getClass() + "): "
            + msg);
        if (msg instanceof BackendTransfer)
        {
          logger.info(getControllerName() + ":Received transfer command");
          BackendTransfer transfer = (BackendTransfer) msg;
          BackendInfo info = transfer.getInfo();
          DatabaseBackend backend = new DatabaseBackend(info);
          try
          {
            this.addBackend(backend);
          }
          catch (Exception e1)
          {
            logger.error("Transfer failed", e1);
            return e1;
          }
          logger.info(getControllerName() + ":Enable backend");
          enableBackend(backend.getName());
          return Boolean.TRUE;
        }
        // Other message types will be handled in multithreaaded handler
      }
      else
      {
        String errorMsg = "Invalid null message";
        logger.error(errorMsg);
        return new ControllerException(errorMsg);
      }
      return null;
    }
    catch (Exception e)
    {
      if (e instanceof RuntimeException)
        logger.warn("Error while handling group message:" + msg.getClass(), e);
      return e;
    }
  }

  /**
   * @see org.objectweb.tribe.adapters.MulticastRequestListener#handleMessageMultiThreaded(Serializable,
   *      Member, Object)
   */
  public Serializable handleMessageMultiThreaded(Serializable msg,
      Member sender, Object handleMessageSingleThreadedResult)
  {
    try
    {
      if (msg != null)
      {
        logger.debug("handleMessageMultiThreaded (" + msg.getClass() + "): "
            + msg);
        if (msg instanceof DistributedRequest)
        { //Distributed request execution
          if (logger.isDebugEnabled())
            logger.debug(getControllerName() + ": DistributedRequest "
                + ((DistributedRequest) msg).getRequest().getId() + " from "
                + sender);
          ((DistributedRequest) msg)
              .scheduleRequest((DistributedRequestManager) this.requestManager);
          return ((Serializable) ((DistributedRequest) msg)
              .executeScheduledRequest((DistributedRequestManager) this.requestManager));
        }
        if (msg instanceof ControllerName)
        {
          String controllerName = ((ControllerName) msg).getJmxName();
          if (logger.isDebugEnabled())
            logger.debug(getControllerName() + ": New ControllerName "
                + controllerName);
          controllersMap.put(sender, controllerName);
          return Boolean.TRUE;
        }
        if (msg instanceof Commit)
        { //Distributed commit execution
          if (logger.isDebugEnabled())
            logger.debug(getControllerName() + ": Commit from " + sender);
          return ((Serializable) ((Commit) msg)
              .commit((DistributedRequestManager) this.requestManager));
        }
        if (msg instanceof Rollback)
        { //Distributed commit execution
          if (logger.isDebugEnabled())
            logger.debug(getControllerName() + ": Rollback from " + sender);
          return ((Serializable) ((Rollback) msg)
              .rollback((DistributedRequestManager) this.requestManager));
        }
        else if (msg instanceof BackendTransfer)
          return (Serializable) handleMessageSingleThreadedResult;
        else if (msg instanceof VirtualDatabaseConfiguration)
        { // Check if given configuration is compatible with the local one
          VirtualDatabaseConfiguration vdc = (VirtualDatabaseConfiguration) msg;
          // Send notification
          if (MBeanServerManager.isJmxEnabled())
          {
            Hashtable data = new Hashtable();
            data.put("controllerName", vdc.getControllerName());
            data.put("rmiconnector", new String[]{vdc.getRmiHostname(),
                vdc.getRmiPort()});
            RmiConnector.broadcastNotification(this,
                CjdbcNotificationList.DISTRIBUTED_CONTROLLER_ADDED,
                CjdbcNotificationList.NOTIFICATION_LEVEL_INFO, Translate.get(
                    "notification.distributed.controller.added", this
                        .getVirtualDatabaseName()), data);
          }
          controllersMap.put(sender, vdc.getControllerJmxName());
          if (logger.isDebugEnabled())
            logger.debug("VirtualDatabaseConfiguration from "
                + vdc.getControllerName());

          // Send controller name to new comer
          ArrayList target = new ArrayList();
          target.add(sender);
          multicastRequestAdapter.multicastMessage(target, new ControllerName(
              controller.getControllerName(), controller.getJmxName()),
              MulticastRequestAdapter.WAIT_ALL,
              CJDBCGroupMessage.defaultCastTimeOut);

          // Broadcast backends
          multicastRequestAdapter.multicastMessage(target, new BackendStatus(
              getBackendsInfo(backends)), MulticastRequestAdapter.WAIT_ALL,
              CJDBCGroupMessage.defaultCastTimeOut);

          if (isLocalSender(sender))
            return Boolean.TRUE;
          else
            return new Boolean((vdc.isCompatible(this)));
        }
        else if (msg instanceof BackendStatus)
        { // Update backend list from sender
          ArrayList remoteBackendInfoList = ((BackendStatus) msg).getBackends();
          // Convert BackendInfo arraylist to real DatabaseBackend objects
          ArrayList remoteBackendList = new ArrayList(remoteBackendInfoList
              .size());
          for (Iterator iter = remoteBackendInfoList.iterator(); iter.hasNext();)
          {
            BackendInfo info = (BackendInfo) iter.next();
            remoteBackendList.add(info.getDatabaseBackend());
          }
          backendsPerController.put(sender, remoteBackendList);
          if (logger.isInfoEnabled())
            logger
                .info(Translate
                    .get(
                        "virtualdatabase.distributed.configuration.updating.backend.list",
                        sender));
        }
        else if (msg instanceof EnableBackend)
        {
          ArrayList remoteBackendList = (ArrayList) backendsPerController
              .get(sender);
          if (remoteBackendList == null)
          { // This case was reported by Alessandro Gamboz on April 1, 2005.
            // It looks like the EnableBackend message arrives before membership
            // has been properly updated.
            logger.warn("No information has been found for remote controller "
                + sender);
            remoteBackendList = new ArrayList();
            backendsPerController.put(sender, remoteBackendList);
          }
          DatabaseBackend enabledBackend = ((EnableBackend) msg)
              .getDatabaseBackend();
          int size = remoteBackendList.size();
          boolean backendFound = false;
          for (int i = 0; i < size; i++)
          {
            DatabaseBackend b = (DatabaseBackend) remoteBackendList.get(i);
            if (b.equals(enabledBackend))
            {
              logger.info("Backend " + b.getName() + " enabled on controller "
                  + sender);
              remoteBackendList.set(i, enabledBackend);
              backendFound = true;
              break;
            }
          }
          if (!backendFound)
          {
            logger
                .warn("Updating backend list with unknown backend "
                    + enabledBackend.getName() + " enabled on controller "
                    + sender);
            remoteBackendList.add(enabledBackend);
          }
        }
        else
          logger.warn("Unhandled message type received: " + msg.getClass()
              + "(" + msg + ")");
      }
      else
      {
        String errorMsg = "Invalid null message";
        logger.error(errorMsg);
        return new ControllerException(errorMsg);
      }
      return null;
    }
    catch (Exception e)
    {
      if (e instanceof RuntimeException)
        logger.warn("Error while handling group message:" + msg.getClass(), e);
      return e;
    }
  }

  /**
   * Returns true if the given member is ourselves.
   * 
   * @param sender the sender
   * @return true if we are the sender, false otherwise
   */
  private boolean isLocalSender(Member sender)
  {
    return channel.getLocalMembership().equals(sender);
  }

  /**
   * Get the status of all remote controllers
   * 
   * @throws NotConnectedException if the channel is not connected
   * @throws ChannelException if the channel reported an error
   * @throws TimeoutException if a timeout occured
   */
  public void getBackendStatus() throws TimeoutException, ChannelException,
      NotConnectedException
  {
    if (logger.isDebugEnabled())
      logger.debug("Requesting remote controllers status");
    MulticastResponse rspList = multicastRequestAdapter.multicastMessage(null,
        new BackendStatus(getBackendsInfo(backends)),
        MulticastRequestAdapter.WAIT_ALL, CJDBCGroupMessage.defaultCastTimeOut);

    HashMap results = rspList.getResults();
    for (Iterator iter = results.values().iterator(); iter.hasNext();)
    {
      ArrayList b = (ArrayList) iter.next();
      int bSize = b.size();
      if (bSize == 0)
        logger.debug("No Database backends");
      else
        for (int j = 0; j < bSize; j++)
          logger.debug(((DatabaseBackend) b.get(j)).getXml());
    }
  }

  //
  // Membership support
  //

  /**
   * @see org.jgroups.MembershipListener#viewAccepted(org.jgroups.View)
   */
  public void viewAccepted(Group newGroup)
  {
    ArrayList newMembers = newGroup.getMembers();
    int newSize = newMembers.size();
    for (int i = 0; i < newSize; i++)
    {
      Member m = (Member) newMembers.get(i);
      if (!currentGroup.hasMember(m))
      {
        logger.info(controller.getControllerName() + ":New controller " + m.toString()
            + " detected.");
      }
    }
    ArrayList currentMembers = currentGroup.getMembers();
    logger.info(controller.getControllerName() + ":Current Members:" + currentMembers);
    int currentSize = currentMembers.size();
    for (int i = 0; i < currentSize; i++)
    {
      Member m = (Member) currentMembers.get(i);
      if (!newGroup.hasMember(m))
      {
        logger.warn("Controller " + m + " has failed.");
      }
    }
    currentGroup = newGroup;
    currentMembers = currentGroup.getMembers();
    logger.info(controller.getControllerName() + ":new Members:" + currentMembers);
    allMemberButUs = ((ArrayList) currentGroup.getMembers().clone());
    allMemberButUs.remove(channel.getLocalMembership());
  }

  //
  // Getter/Setter and tools (equals, ...)
  //

  /**
   * Get the backend distribution policies
   * 
   * @return an ArrayList of BackendRecoveryPolicy objects
   */
  public HashMap getBackendRecoveryPolicy()
  {
    return backendRecoveryPolicy;
  }

  /**
   * Is this virtual database distributed ?
   * 
   * @return true
   */
  public boolean isDistributed()
  {
    return true;
  }

  /**
   * Two virtual databases are equal if they have the same name, login and
   * password.
   * 
   * @param other an object
   * @return a <code>boolean</code> value
   */
  public boolean equals(Object other)
  {
    if ((other == null)
        || (!(other instanceof org.objectweb.cjdbc.controller.virtualdatabase.DistributedVirtualDatabase)))
      return false;
    else
    {
      DistributedVirtualDatabase db = (org.objectweb.cjdbc.controller.virtualdatabase.DistributedVirtualDatabase) other;
      return name.equals(db.getDatabaseName())
          && groupName.equals(db.getGroupName());
    }
  }

  /**
   * Get the XML dump of the Distribution element if any.
   * 
   * @return XML dump of the Distribution element
   */
  protected String getDistributionXml()
  {
    StringBuffer info = new StringBuffer();
    info.append("<" + DatabasesXmlTags.ELT_Distribution + " "
        + DatabasesXmlTags.ATT_groupName + "=\"" + groupName + "\">");

    for (Iterator iter = backendRecoveryPolicy.values().iterator(); iter
        .hasNext();)
    {
      BackendRecoveryPolicy p = (BackendRecoveryPolicy) iter.next();
      info.append(p.getXml());
    }

    info.append("</" + DatabasesXmlTags.ELT_Distribution + ">");
    return info.toString();
  }

  /**
   * @see org.objectweb.cjdbc.common.jmx.mbeans.VirtualDatabaseMBean#viewControllerList()
   */
  public String[] viewControllerList()
  {
    if (logger.isInfoEnabled())
    {
      logger.info(channel.getLocalMembership() + " see members:"
          + currentGroup.getMembers() + " and has mapping:" + controllersMap);
    }
    String[] members = new String[controllersMap.keySet().size()];
    Iterator iter = controllersMap.keySet().iterator();
    int i = 0;
    while (iter.hasNext())
    {
      members[i] = (String) controllersMap.get(iter.next());
      i++;
    }
    return members;
  }

  /**
   * (non-Javadoc)
   * 
   * @see org.objectweb.cjdbc.controller.virtualdatabase.VirtualDatabase#setAuthenticationManager(org.objectweb.cjdbc.controller.authentication.AuthenticationManager)
   */
  public void setAuthenticationManager(
      AuthenticationManager authenticationManager)
  {
    super.setAuthenticationManager(authenticationManager);
  }

  /**
   * @see org.objectweb.cjdbc.common.jmx.mbeans.VirtualDatabaseMBean#viewGroupBackends()
   */
  public Hashtable viewGroupBackends() throws VirtualDatabaseException
  {
    Hashtable map = new Hashtable(controllersMap.size());
    Iterator iter = backendsPerController.keySet().iterator();
    Address addr;
    while (iter.hasNext())
    {
      addr = (Address) iter.next();
      map.put(controllersMap.get(addr), backendsPerController.get(addr));
    }
    return map;
  }

  /**
   * We have to convert the backends list from an array of
   * <code>DatabaseBackend</code> object to an ArrayList of
   * <code>BackendInfo</code> objects. The DatabaseBackend objects cannot be
   * serialized because they are used as MBean and notification emitters, so we
   * want to extract the info out of them.
   * 
   * @param backendsObject the list of DatabaseBackend object
   * @see BackendInfo
   * @return a list of BackendInfo objects
   */
  public ArrayList getBackendsInfo(ArrayList backendsObject)
  {
    int size = backendsObject.size();
    ArrayList infos = new ArrayList(size);
    DatabaseBackend backend;
    for (int i = 0; i < size; i++)
    {
      backend = (DatabaseBackend) backendsObject.get(i);
      infos.add(createBackendInfo(backend, false));
    }
    return infos;
  }

  /**
   * Create backend information object from a DatabaseBackend object This will
   * get only static information
   * 
   * @param backend the <code>DatabaseBackend</code> object to get info from
   * @param useXml should we use xml for extensive backend description
   * @return <code>BackendInfo</code>
   */
  public BackendInfo createBackendInfo(DatabaseBackend backend, boolean useXml)
  {
    if (useXml)
      try
      {
        return new BackendInfo(backend.getXml());
      }
      catch (DocumentException e)
      {
        // go to next method. cause we haven't returned yet
      }
    return new BackendInfo(backend);
  }

  /**
   * @see org.objectweb.cjdbc.common.jmx.mbeans.VirtualDatabaseMBean#removeBackend(java.lang.String)
   */
  public void removeBackend(String backend) throws VirtualDatabaseException
  {
    super.removeBackend(backend);

    try
    {
      // Send a group message to update backend list
      broadcastBackendInformation(allMemberButUs);
    }
    catch (Exception e)
    {
      String msg = "An error occured while multicasting new backedn information";
      logger.error(msg, e);
      throw new VirtualDatabaseException(msg, e);
    }
  }

  /**
   * @see org.objectweb.cjdbc.common.jmx.mbeans.VirtualDatabaseMBean#transferBackend(java.lang.String,
   *      java.lang.String)
   */
  public void transferBackend(String backend, String controllerDestination)
      throws VirtualDatabaseException
  {
    // Get the target controller
    ArrayList dest = new ArrayList(1);
    Iterator iter = controllersMap.entrySet().iterator();
    Entry entry;
    Member targetMember = null;
    while (iter.hasNext())
    {
      entry = (Entry) iter.next();
      if (entry.getValue().equals(controllerDestination))
        targetMember = (Member) entry.getKey();
    }
    if (targetMember == null)
      throw new VirtualDatabaseException("Cannot find controller:"
          + controllerDestination + " in group");
    dest.add(targetMember);

    // Get reference on backend
    DatabaseBackend db = getAndCheckBackend(backend, NO_CHECK_BACKEND);

    //  Disable backend
    try
    {
      DistributedRequestManager manager = ((DistributedRequestManager) requestManager);
      if (hasRecoveryLog())
        manager.disableBackendForCheckpoint(db, "transfer of " + backend
            + " to :" + controllerDestination);
      else
        manager.disableBackend(db);
    }
    catch (SQLException e)
    {
      throw new VirtualDatabaseException(e.getMessage());
    }

    try
    {
      //  Send backend transfer message
      if (logger.isDebugEnabled())
        logger.debug("Sending transfer message to:" + targetMember);

      multicastRequestAdapter.multicastMessage(dest, new BackendTransfer(
          controllerDestination, createBackendInfo(db, true)),
          MulticastRequestAdapter.WAIT_ALL,
          CJDBCGroupMessage.defaultCastTimeOut);

      // Remove backend from this controller
      removeBackend(db);

      // Broadcast updated backend list
      broadcastBackendInformation(allMemberButUs);
    }
    catch (Exception e)
    {
      String msg = "An error occured while transfering the backend";
      logger.error(msg, e);
      throw new VirtualDatabaseException(msg, e);
    }
  }

}

---