Metacat

/**

 *  '$RCSfile$'

 *  Copyright: 2000-2011 Regents of the University of California and the

 *              National Center for Ecological Analysis and Synthesis

 *

 *   '$Author:  $'

 *     '$Date:  $'

 *

 * This program is free software; you can redistribute it and/or modify

 * it under the terms of the GNU General Public License as published by

 * the Free Software Foundation; either version 2 of the License, or

 * (at your option) any later version.

 *

 * This program 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 General Public License for more details.

 *

 * You should have received a copy of the GNU General Public License

 * along with this program; if not, write to the Free Software

 * Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA

 */

package edu.ucsb.nceas.metacat.dataone;

import java.io.InputStream;

import java.math.BigInteger;

import java.util.ArrayList;

import java.util.Calendar;

import java.util.Date;

import java.util.List;

import java.util.concurrent.locks.Lock;

import javax.servlet.http.HttpServletRequest;

import org.apache.log4j.Logger;

import org.dataone.client.v2.CNode;

import org.dataone.client.v2.itk.D1Client;

import org.dataone.client.v2.MNode;

import org.dataone.service.cn.v2.CNAuthorization;

import org.dataone.service.cn.v2.CNCore;

import org.dataone.service.cn.v2.CNRead;

import org.dataone.service.cn.v2.CNReplication;

import org.dataone.service.exceptions.BaseException;

import org.dataone.service.exceptions.IdentifierNotUnique;

import org.dataone.service.exceptions.InsufficientResources;

import org.dataone.service.exceptions.InvalidRequest;

import org.dataone.service.exceptions.InvalidSystemMetadata;

import org.dataone.service.exceptions.InvalidToken;

import org.dataone.service.exceptions.NotAuthorized;

import org.dataone.service.exceptions.NotFound;

import org.dataone.service.exceptions.NotImplemented;

import org.dataone.service.exceptions.ServiceFailure;

import org.dataone.service.exceptions.UnsupportedType;

import org.dataone.service.exceptions.VersionMismatch;

import org.dataone.service.types.v1.AccessPolicy;

import org.dataone.service.types.v1.Checksum;

import org.dataone.service.types.v1.ChecksumAlgorithmList;

import org.dataone.service.types.v1.DescribeResponse;

import org.dataone.service.types.v1.Event;

import org.dataone.service.types.v1.Identifier;

import org.dataone.service.types.v2.Log;

import org.dataone.service.types.v2.Node;

import org.dataone.service.types.v2.NodeList;

import org.dataone.service.types.v1.NodeReference;

import org.dataone.service.types.v1.NodeType;

import org.dataone.service.types.v2.ObjectFormat;

import org.dataone.service.types.v1.ObjectFormatIdentifier;

import org.dataone.service.types.v2.ObjectFormatList;

import org.dataone.service.types.v1.ObjectList;

import org.dataone.service.types.v1.ObjectLocationList;

import org.dataone.service.types.v1.Permission;

import org.dataone.service.types.v1.Replica;

import org.dataone.service.types.v1.ReplicationPolicy;

import org.dataone.service.types.v1.ReplicationStatus;

import org.dataone.service.types.v1.Session;

import org.dataone.service.types.v1.Subject;

import org.dataone.service.types.v2.SystemMetadata;

import org.dataone.service.types.v2.util.ServiceMethodRestrictionUtil;

import org.dataone.service.types.v1_1.QueryEngineDescription;

import org.dataone.service.types.v1_1.QueryEngineList;

import edu.ucsb.nceas.metacat.EventLog;

import edu.ucsb.nceas.metacat.IdentifierManager;

import edu.ucsb.nceas.metacat.McdbDocNotFoundException;

import edu.ucsb.nceas.metacat.dataone.hazelcast.HazelcastService;

import edu.ucsb.nceas.metacat.index.MetacatSolrIndex;

/**

 * Represents Metacat's implementation of the DataONE Coordinating Node 

 * service API. Methods implement the various CN* interfaces, and methods common

 * to both Member Node and Coordinating Node interfaces are found in the

 * D1NodeService super class.

 *

 */

public class CNodeService extends D1NodeService implements CNAuthorization,

    CNCore, CNRead, CNReplication {

  /* the logger instance */

  private Logger logMetacat = null;

  /**

   * singleton accessor

   */

  public static CNodeService getInstance(HttpServletRequest request) { 

    return new CNodeService(request);

  }

  /**

   * Constructor, private for singleton access

   */

  private CNodeService(HttpServletRequest request) {

    super(request);

    logMetacat = Logger.getLogger(CNodeService.class);

  }

  /**

   * Set the replication policy for an object given the object identifier

   * 

   * @param session - the Session object containing the credentials for the Subject

   * @param pid - the object identifier for the given object

   * @param policy - the replication policy to be applied

   * 

   * @return true or false

   * 

   * @throws NotImplemented

   * @throws NotAuthorized

   * @throws ServiceFailure

   * @throws InvalidRequest

   * @throws VersionMismatch

   * 

   */

  @Override

  public boolean setReplicationPolicy(Session session, Identifier pid,

      ReplicationPolicy policy, long serialVersion) 

      throws NotImplemented, NotFound, NotAuthorized, ServiceFailure, 

      InvalidRequest, InvalidToken, VersionMismatch {

      // The lock to be used for this identifier

      Lock lock = null;

      // get the subject

      Subject subject = session.getSubject();

      // are we allowed to do this?

      if (!isAuthorized(session, pid, Permission.CHANGE_PERMISSION)) {

          throw new NotAuthorized("4881", Permission.CHANGE_PERMISSION

                  + " not allowed by " + subject.getValue() + " on "

                  + pid.getValue());

      }

      SystemMetadata systemMetadata = null;

      try {

          lock = HazelcastService.getInstance().getLock(pid.getValue());

          lock.lock();

          logMetacat.debug("Locked identifier " + pid.getValue());

          try {

              if ( HazelcastService.getInstance().getSystemMetadataMap().containsKey(pid) ) {

                  systemMetadata = HazelcastService.getInstance().getSystemMetadataMap().get(pid);

              }

              // did we get it correctly?

              if ( systemMetadata == null ) {

                  throw new NotFound("4884", "Couldn't find an object identified by " + pid.getValue());

              }

              // does the request have the most current system metadata?

              if ( systemMetadata.getSerialVersion().longValue() != serialVersion ) {

                 String msg = "The requested system metadata version number " + 

                     serialVersion + " differs from the current version at " +

                     systemMetadata.getSerialVersion().longValue() +

                     ". Please get the latest copy in order to modify it.";

                 throw new VersionMismatch("4886", msg);

              }

          } catch (RuntimeException e) { // Catch is generic since HZ throws RuntimeException

              throw new NotFound("4884", "No record found for: " + pid.getValue());

          }

          // set the new policy

          systemMetadata.setReplicationPolicy(policy);

          // update the metadata

          try {

              systemMetadata.setSerialVersion(systemMetadata.getSerialVersion().add(BigInteger.ONE));

              systemMetadata.setDateSysMetadataModified(Calendar.getInstance().getTime());

              HazelcastService.getInstance().getSystemMetadataMap().put(systemMetadata.getIdentifier(), systemMetadata);

              notifyReplicaNodes(systemMetadata);

          } catch (RuntimeException e) {

              throw new ServiceFailure("4882", e.getMessage());

          }

      } catch (RuntimeException e) {

          throw new ServiceFailure("4882", e.getMessage());

      } finally {

          lock.unlock();

          logMetacat.debug("Unlocked identifier " + pid.getValue());

      }

      return true;

  }

  /**

   * Deletes the replica from the given Member Node

   * NOTE: MN.delete() may be an "archive" operation. TBD.

   * @param session

   * @param pid

   * @param nodeId

   * @param serialVersion

   * @return

   * @throws InvalidToken

   * @throws ServiceFailure

   * @throws NotAuthorized

   * @throws NotFound

   * @throws NotImplemented

   * @throws VersionMismatch

   */

  @Override

  public boolean deleteReplicationMetadata(Session session, Identifier pid, NodeReference nodeId, long serialVersion) 

  	throws InvalidToken, ServiceFailure, NotAuthorized, NotFound, NotImplemented, VersionMismatch {

	  	// The lock to be used for this identifier

		Lock lock = null;

		// get the subject

		Subject subject = session.getSubject();

		// are we allowed to do this?

		boolean isAuthorized = false;

		try {

			isAuthorized = isAuthorized(session, pid, Permission.WRITE);

		} catch (InvalidRequest e) {

			throw new ServiceFailure("4882", e.getDescription());

		}

		if (!isAuthorized) {

			throw new NotAuthorized("4881", Permission.WRITE

					+ " not allowed by " + subject.getValue() + " on "

					+ pid.getValue());

		}

		SystemMetadata systemMetadata = null;

		try {

			lock = HazelcastService.getInstance().getLock(pid.getValue());

			lock.lock();

			logMetacat.debug("Locked identifier " + pid.getValue());

			try {

				if (HazelcastService.getInstance().getSystemMetadataMap().containsKey(pid)) {

					systemMetadata = HazelcastService.getInstance().getSystemMetadataMap().get(pid);

				}

				// did we get it correctly?

				if (systemMetadata == null) {

					throw new NotFound("4884", "Couldn't find an object identified by " + pid.getValue());

				}

				// does the request have the most current system metadata?

				if (systemMetadata.getSerialVersion().longValue() != serialVersion) {

					String msg = "The requested system metadata version number "

							+ serialVersion

							+ " differs from the current version at "

							+ systemMetadata.getSerialVersion().longValue()

							+ ". Please get the latest copy in order to modify it.";

					throw new VersionMismatch("4886", msg);

				}

			} catch (RuntimeException e) { // Catch is generic since HZ throws RuntimeException

				throw new NotFound("4884", "No record found for: " + pid.getValue());

			}

			// check permissions

			// TODO: is this necessary?

			List<Node> nodeList = D1Client.getCN().listNodes().getNodeList();

			boolean isAllowed = ServiceMethodRestrictionUtil.isMethodAllowed(session.getSubject(), nodeList, "CNReplication", "deleteReplicationMetadata");

			if (!isAllowed) {

				throw new NotAuthorized("4881", "Caller is not authorized to deleteReplicationMetadata");

			}

			// delete the replica from the given node

			// CSJ: use CN.delete() to truly delete a replica, semantically

			// deleteReplicaMetadata() only modifies the sytem metadata entry.

			//D1Client.getMN(nodeId).delete(session, pid);

			// reflect that change in the system metadata

			List<Replica> updatedReplicas = new ArrayList<Replica>(systemMetadata.getReplicaList());

			for (Replica r: systemMetadata.getReplicaList()) {

				  if (r.getReplicaMemberNode().equals(nodeId)) {

					  updatedReplicas.remove(r);

					  break;

				  }

			}

			systemMetadata.setReplicaList(updatedReplicas);

			// update the metadata

			try {

				systemMetadata.setSerialVersion(systemMetadata.getSerialVersion().add(BigInteger.ONE));

				systemMetadata.setDateSysMetadataModified(Calendar.getInstance().getTime());

				HazelcastService.getInstance().getSystemMetadataMap().put(systemMetadata.getIdentifier(), systemMetadata);

			} catch (RuntimeException e) {

				throw new ServiceFailure("4882", e.getMessage());

			}

		} catch (RuntimeException e) {

			throw new ServiceFailure("4882", e.getMessage());

		} finally {

			lock.unlock();

			logMetacat.debug("Unlocked identifier " + pid.getValue());

		}

		return true;	  

  }

  /**

   * Deletes an object from the Coordinating Node

   * 

   * @param session - the Session object containing the credentials for the Subject

   * @param pid - The object identifier to be deleted

   * 

   * @return pid - the identifier of the object used for the deletion

   * 

   * @throws InvalidToken

   * @throws ServiceFailure

   * @throws NotAuthorized

   * @throws NotFound

   * @throws NotImplemented

   * @throws InvalidRequest

   */

  @Override

  public Identifier delete(Session session, Identifier pid) 

      throws InvalidToken, ServiceFailure, NotAuthorized, NotFound, NotImplemented {

      String localId = null;      // The corresponding docid for this pid

	  Lock lock = null;           // The lock to be used for this identifier

      CNode cn = null;            // a reference to the CN to get the node list    

      NodeType nodeType = null;   // the nodeType of the replica node being contacted

      List<Node> nodeList = null; // the list of nodes in this CN environment

      // check for a valid session

      if (session == null) {

        	throw new InvalidToken("4963", "No session has been provided");

      }

      // do we have a valid pid?

      if (pid == null || pid.getValue().trim().equals("")) {

          throw new ServiceFailure("4960", "The provided identifier was invalid.");

      }

	  // check that it is CN/admin

	  boolean allowed = isAdminAuthorized(session);

	  // additional check if it is the authoritative node if it is not the admin

      if(!allowed) {

          allowed = isAuthoritativeMNodeAdmin(session, pid);

      }

	  if (!allowed) {

		  String msg = "The subject " + session.getSubject().getValue() + 

			  " is not allowed to call delete() on a Coordinating Node.";

		  logMetacat.info(msg);

		  throw new NotAuthorized("4960", msg);

	  }

	  // Don't defer to superclass implementation without a locally registered identifier

	  SystemMetadata systemMetadata = HazelcastService.getInstance().getSystemMetadataMap().get(pid);

      // Check for the existing identifier

      try {

          localId = IdentifierManager.getInstance().getLocalId(pid.getValue());

          super.delete(session, pid);

      } catch (McdbDocNotFoundException e) {

          // This object is not registered in the identifier table. Assume it is of formatType DATA,

    	  // and set the archive flag. (i.e. the *object* doesn't exist on the CN)

          try {

  			  lock = HazelcastService.getInstance().getLock(pid.getValue());

  			  lock.lock();

  			  logMetacat.debug("Locked identifier " + pid.getValue());

			  SystemMetadata sysMeta = HazelcastService.getInstance().getSystemMetadataMap().get(pid);

			  if ( sysMeta != null ) {

				/*sysMeta.setSerialVersion(sysMeta.getSerialVersion().add(BigInteger.ONE));

				sysMeta.setArchived(true);

				sysMeta.setDateSysMetadataModified(Calendar.getInstance().getTime());

				HazelcastService.getInstance().getSystemMetadataMap().put(pid, sysMeta);*/

			    //move the systemmetadata object from the map and delete the records in the systemmetadata database table

	            //since this is cn, we don't need worry about the mn solr index.

	            HazelcastService.getInstance().getSystemMetadataMap().remove(pid);

	            HazelcastService.getInstance().getIdentifiers().remove(pid);

			  } else {

				  throw new ServiceFailure("4962", "Couldn't delete the object " + pid.getValue() +

					  ". Couldn't obtain the system metadata record.");

			  }

		  } catch (RuntimeException re) {

			  throw new ServiceFailure("4962", "Couldn't delete " + pid.getValue() + 

				  ". The error message was: " + re.getMessage());

		  } finally {

			  lock.unlock();

			  logMetacat.debug("Unlocked identifier " + pid.getValue());

		  }

          // NOTE: cannot log the delete without localId

//          EventLog.getInstance().log(request.getRemoteAddr(), 

//                  request.getHeader("User-Agent"), session.getSubject().getValue(), 

//                  pid.getValue(), Event.DELETE.xmlValue());

      }

      // get the node list

      try {

          cn = D1Client.getCN();

          nodeList = cn.listNodes().getNodeList();

      } catch (Exception e) { // handle BaseException and other I/O issues

          // swallow errors since the call is not critical

          logMetacat.error("Can't inform MNs of the deletion of " + pid.getValue() + 

              " due to communication issues with the CN: " + e.getMessage());

      }

	  // notify the replicas

	  if (systemMetadata.getReplicaList() != null) {

		  for (Replica replica: systemMetadata.getReplicaList()) {

			  NodeReference replicaNode = replica.getReplicaMemberNode();

			  try {

                  if (nodeList != null) {

                      // find the node type

                      for (Node node : nodeList) {

                          if ( node.getIdentifier().getValue().equals(replicaNode.getValue()) ) {

                              nodeType = node.getType();

                              break;

                          }

                      }

                  }

                  // only send call MN.delete() to avoid an infinite loop with the CN

                  if (nodeType != null && nodeType == NodeType.MN) {

				      Identifier mnRetId = D1Client.getMN(replicaNode).delete(null, pid);

                  }

			  } catch (Exception e) {

				  // all we can really do is log errors and carry on with life

				  logMetacat.error("Error deleting pid: " +  pid.getValue() + 

					  " from replica MN: " + replicaNode.getValue(), e);

			}

		  }

	  }

	  return pid;

  }

  /**

   * Archives an object from the Coordinating Node

   * 

   * @param session - the Session object containing the credentials for the Subject

   * @param pid - The object identifier to be deleted

   * 

   * @return pid - the identifier of the object used for the deletion

   * 

   * @throws InvalidToken

   * @throws ServiceFailure

   * @throws NotAuthorized

   * @throws NotFound

   * @throws NotImplemented

   * @throws InvalidRequest

   */

  @Override

  public Identifier archive(Session session, Identifier pid) 

      throws InvalidToken, ServiceFailure, NotAuthorized, NotFound, NotImplemented {

      String localId = null; // The corresponding docid for this pid

	  Lock lock = null;      // The lock to be used for this identifier

      CNode cn = null;            // a reference to the CN to get the node list    

      NodeType nodeType = null;   // the nodeType of the replica node being contacted

      List<Node> nodeList = null; // the list of nodes in this CN environment

      // check for a valid session

      if (session == null) {

        	throw new InvalidToken("4973", "No session has been provided");

      }

      // do we have a valid pid?

      if (pid == null || pid.getValue().trim().equals("")) {

          throw new ServiceFailure("4972", "The provided identifier was invalid.");

      }

	  // check that it is CN/admin

	  boolean allowed = isAdminAuthorized(session);

	  //check if it is the authoritative member node

	  if(!allowed) {

	      allowed = isAuthoritativeMNodeAdmin(session, pid);

	  }

	  if (!allowed) {

		  String msg = "The subject " + session.getSubject().getValue() + 

				  " is not allowed to call archive() on a Coordinating Node.";

		  logMetacat.info(msg);

		  throw new NotAuthorized("4970", msg);

	  }

      // Check for the existing identifier

      try {

          localId = IdentifierManager.getInstance().getLocalId(pid.getValue());

          super.archive(session, pid);

      } catch (McdbDocNotFoundException e) {

          // This object is not registered in the identifier table. Assume it is of formatType DATA,

    	  // and set the archive flag. (i.e. the *object* doesn't exist on the CN)

          try {

  			  lock = HazelcastService.getInstance().getLock(pid.getValue());

  			  lock.lock();

  			  logMetacat.debug("Locked identifier " + pid.getValue());

			  SystemMetadata sysMeta = HazelcastService.getInstance().getSystemMetadataMap().get(pid);

			  if ( sysMeta != null ) {

				sysMeta.setSerialVersion(sysMeta.getSerialVersion().add(BigInteger.ONE));

				sysMeta.setArchived(true);

				sysMeta.setDateSysMetadataModified(Calendar.getInstance().getTime());

				HazelcastService.getInstance().getSystemMetadataMap().put(pid, sysMeta);

			    // notify the replicas

				notifyReplicaNodes(sysMeta);

			  } else {

				  throw new ServiceFailure("4972", "Couldn't archive the object " + pid.getValue() +

					  ". Couldn't obtain the system metadata record.");

			  }

		  } catch (RuntimeException re) {

			  throw new ServiceFailure("4972", "Couldn't archive " + pid.getValue() + 

				  ". The error message was: " + re.getMessage());

		  } finally {

			  lock.unlock();

			  logMetacat.debug("Unlocked identifier " + pid.getValue());

		  }

          // NOTE: cannot log the archive without localId

//          EventLog.getInstance().log(request.getRemoteAddr(), 

//                  request.getHeader("User-Agent"), session.getSubject().getValue(), 

//                  pid.getValue(), Event.DELETE.xmlValue());

      }

	  return pid;

  }

  /**

   * Set the obsoletedBy attribute in System Metadata

   * @param session

   * @param pid

   * @param obsoletedByPid

   * @param serialVersion

   * @return

   * @throws NotImplemented

   * @throws NotFound

   * @throws NotAuthorized

   * @throws ServiceFailure

   * @throws InvalidRequest

   * @throws InvalidToken

   * @throws VersionMismatch

   */

  @Override

  public boolean setObsoletedBy(Session session, Identifier pid,

			Identifier obsoletedByPid, long serialVersion)

			throws NotImplemented, NotFound, NotAuthorized, ServiceFailure,

			InvalidRequest, InvalidToken, VersionMismatch {

		// The lock to be used for this identifier

		Lock lock = null;

		// get the subject

		Subject subject = session.getSubject();

		// are we allowed to do this?

		if (!isAuthorized(session, pid, Permission.WRITE)) {

			throw new NotAuthorized("4881", Permission.WRITE

					+ " not allowed by " + subject.getValue() + " on "

					+ pid.getValue());

		}

		SystemMetadata systemMetadata = null;

		try {

			lock = HazelcastService.getInstance().getLock(pid.getValue());

			lock.lock();

			logMetacat.debug("Locked identifier " + pid.getValue());

			try {

				if (HazelcastService.getInstance().getSystemMetadataMap().containsKey(pid)) {

					systemMetadata = HazelcastService.getInstance().getSystemMetadataMap().get(pid);

				}

				// did we get it correctly?

				if (systemMetadata == null) {

					throw new NotFound("4884", "Couldn't find an object identified by " + pid.getValue());

				}

				// does the request have the most current system metadata?

				if (systemMetadata.getSerialVersion().longValue() != serialVersion) {

					String msg = "The requested system metadata version number "

							+ serialVersion

							+ " differs from the current version at "

							+ systemMetadata.getSerialVersion().longValue()

							+ ". Please get the latest copy in order to modify it.";

					throw new VersionMismatch("4886", msg);

				}

			} catch (RuntimeException e) { // Catch is generic since HZ throws RuntimeException

				throw new NotFound("4884", "No record found for: " + pid.getValue());

			}

			// set the new policy

			systemMetadata.setObsoletedBy(obsoletedByPid);

			// update the metadata

			try {

				systemMetadata.setSerialVersion(systemMetadata.getSerialVersion().add(BigInteger.ONE));

				systemMetadata.setDateSysMetadataModified(Calendar.getInstance().getTime());

				HazelcastService.getInstance().getSystemMetadataMap().put(systemMetadata.getIdentifier(), systemMetadata);

			} catch (RuntimeException e) {

				throw new ServiceFailure("4882", e.getMessage());

			}

		} catch (RuntimeException e) {

			throw new ServiceFailure("4882", e.getMessage());

		} finally {

			lock.unlock();

			logMetacat.debug("Unlocked identifier " + pid.getValue());

		}

		return true;

	}

  /**

   * Set the replication status for an object given the object identifier

   * 

   * @param session - the Session object containing the credentials for the Subject

   * @param pid - the object identifier for the given object

   * @param status - the replication status to be applied

   * 

   * @return true or false

   * 

   * @throws NotImplemented

   * @throws NotAuthorized

   * @throws ServiceFailure

   * @throws InvalidRequest

   * @throws InvalidToken

   * @throws NotFound

   * 

   */

  @Override

  public boolean setReplicationStatus(Session session, Identifier pid,

      NodeReference targetNode, ReplicationStatus status, BaseException failure) 

      throws ServiceFailure, NotImplemented, InvalidToken, NotAuthorized, 

      InvalidRequest, NotFound {

	  // cannot be called by public

	  if (session == null) {

		  throw new NotAuthorized("4720", "Session cannot be null");

	  }

      // The lock to be used for this identifier

      Lock lock = null;

      boolean allowed = false;

      int replicaEntryIndex = -1;

      List<Replica> replicas = null;

      // get the subject

      Subject subject = session.getSubject();

      logMetacat.debug("ReplicationStatus for identifier " + pid.getValue() +

          " is " + status.toString());

      SystemMetadata systemMetadata = null;

      try {

          lock = HazelcastService.getInstance().getLock(pid.getValue());

          lock.lock();

          logMetacat.debug("Locked identifier " + pid.getValue());

          try {      

              systemMetadata = HazelcastService.getInstance().getSystemMetadataMap().get(pid);

              // did we get it correctly?

              if ( systemMetadata == null ) {

                  logMetacat.debug("systemMetadata is null for " + pid.getValue());

                  throw new NotFound("4740", "Couldn't find an object identified by " + pid.getValue());

              }

              replicas = systemMetadata.getReplicaList();

              int count = 0;

              // was there a failure? log it

              if ( failure != null && status.equals(ReplicationStatus.FAILED) ) {

                 String msg = "The replication request of the object identified by " + 

                     pid.getValue() + " failed.  The error message was " +

                     failure.getMessage() + ".";

              }

              if (replicas.size() > 0 && replicas != null) {

                  // find the target replica index in the replica list

                  for (Replica replica : replicas) {

                      String replicaNodeStr = replica.getReplicaMemberNode().getValue();

                      String targetNodeStr = targetNode.getValue();

                      logMetacat.debug("Comparing " + replicaNodeStr + " to " + targetNodeStr);

                      if (replicaNodeStr.equals(targetNodeStr)) {

                          replicaEntryIndex = count;

                          logMetacat.debug("replica entry index is: "

                                  + replicaEntryIndex);

                          break;

                      }

                      count++;

                  }

              }

              // are we allowed to do this? only CNs and target MNs are allowed

              CNode cn = D1Client.getCN();

              List<Node> nodes = cn.listNodes().getNodeList();

              // find the node in the node list

              for ( Node node : nodes ) {

                  NodeReference nodeReference = node.getIdentifier();

                  logMetacat.debug("In setReplicationStatus(), Node reference is: " + 

                      nodeReference.getValue());

                  // allow target MN certs

                  if ( targetNode.getValue().equals(nodeReference.getValue() ) &&

                      node.getType().equals(NodeType.MN)) {

                      List<Subject> nodeSubjects = node.getSubjectList();

                      // check if the session subject is in the node subject list

                      for (Subject nodeSubject : nodeSubjects) {

                          logMetacat.debug("In setReplicationStatus(), comparing subjects: " +

                                  nodeSubject.getValue() + " and " + subject.getValue());

                          if ( nodeSubject.equals(subject) ) { // subject of session == target node subject

                              // lastly limit to COMPLETED, INVALIDATED,

                              // and FAILED status updates from MNs only

                              if ( status.equals(ReplicationStatus.COMPLETED) ||

                                   status.equals(ReplicationStatus.INVALIDATED) ||

                                   status.equals(ReplicationStatus.FAILED)) {

                                  allowed = true;

                                  break;

                              }                              

                          }

                      }                 

                  }

              }

              if ( !allowed ) {

                  //check for CN admin access

                  allowed = isAuthorized(session, pid, Permission.WRITE);

              }              

              if ( !allowed ) {

                  String msg = "The subject identified by "

                          + subject.getValue()

                          + " does not have permission to set the replication status for "

                          + "the replica identified by "

                          + targetNode.getValue() + ".";

                  logMetacat.info(msg);

                  throw new NotAuthorized("4720", msg);

              }

          } catch (RuntimeException e) { // Catch is generic since HZ throws RuntimeException

            throw new NotFound("4740", "No record found for: " + pid.getValue() +

                " : " + e.getMessage());

          }

          Replica targetReplica = new Replica();

          // set the status for the replica

          if ( replicaEntryIndex != -1 ) {

              targetReplica = replicas.get(replicaEntryIndex);

              // don't allow status to change from COMPLETED to anything other

              // than INVALIDATED: prevents overwrites from race conditions

              if ( targetReplica.getReplicationStatus().equals(ReplicationStatus.COMPLETED) &&

            	   !status.equals(ReplicationStatus.INVALIDATED)) {

            	  throw new InvalidRequest("4730", "Status state change from " +

            			  targetReplica.getReplicationStatus() + " to " +

            			  status.toString() + "is prohibited for identifier " +

            			  pid.getValue() + " and target node " + 

            			  targetReplica.getReplicaMemberNode().getValue());

              }

              targetReplica.setReplicationStatus(status);

              logMetacat.debug("Set the replication status for " + 

                  targetReplica.getReplicaMemberNode().getValue() + " to " +

                  targetReplica.getReplicationStatus() + " for identifier " +

                  pid.getValue());

          } else {

              // this is a new entry, create it

              targetReplica.setReplicaMemberNode(targetNode);

              targetReplica.setReplicationStatus(status);

              targetReplica.setReplicaVerified(Calendar.getInstance().getTime());

              replicas.add(targetReplica);

          }

          systemMetadata.setReplicaList(replicas);

          // update the metadata

          try {

              systemMetadata.setSerialVersion(systemMetadata.getSerialVersion().add(BigInteger.ONE));

              systemMetadata.setDateSysMetadataModified(Calendar.getInstance().getTime());

              HazelcastService.getInstance().getSystemMetadataMap().put(systemMetadata.getIdentifier(), systemMetadata);

              if ( !status.equals(ReplicationStatus.QUEUED) && 

            	   !status.equals(ReplicationStatus.REQUESTED)) {

                logMetacat.trace("METRICS:\tREPLICATION:\tEND REQUEST:\tPID:\t" + pid.getValue() + 

                          "\tNODE:\t" + targetNode.getValue() + 

                          "\tSIZE:\t" + systemMetadata.getSize().intValue());

                logMetacat.trace("METRICS:\tREPLICATION:\t" + status.toString().toUpperCase() +

                          "\tPID:\t"  + pid.getValue() + 

                          "\tNODE:\t" + targetNode.getValue() + 

                          "\tSIZE:\t" + systemMetadata.getSize().intValue());

              }

              if ( status.equals(ReplicationStatus.FAILED) && failure != null ) {

                  logMetacat.warn("Replication failed for identifier " + pid.getValue() +

                      " on target node " + targetNode + ". The exception was: " +

                      failure.getMessage());

              }

			  // update the replica nodes about the completed replica when complete

              if (status.equals(ReplicationStatus.COMPLETED)) {

				notifyReplicaNodes(systemMetadata);

			}

          } catch (RuntimeException e) {

              throw new ServiceFailure("4700", e.getMessage());

          }

    } catch (RuntimeException e) {

        String msg = "There was a RuntimeException getting the lock for " +

            pid.getValue();

        logMetacat.info(msg);

    } finally {

        lock.unlock();

        logMetacat.debug("Unlocked identifier " + pid.getValue());

    }

      return true;

  }

/**

   * Return the checksum of the object given the identifier 

   * 

   * @param session - the Session object containing the credentials for the Subject

   * @param pid - the object identifier for the given object

   * 

   * @return checksum - the checksum of the object

   * 

   * @throws InvalidToken

   * @throws ServiceFailure

   * @throws NotAuthorized

   * @throws NotFound

   * @throws NotImplemented

   */