/src/edu/ucsb/nceas/metacat/dataone/CNodeService.java - Annotate - Metacat - Ecoinformatics Redmine

metacat/src/edu/ucsb/nceas/metacat/dataone/CNodeService.java @ 9391

-            cjones
+/**
  *  '$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;
-            tao
+import java.io.IOException;
-            cjones
+import java.io.InputStream;
-            tao
+import java.io.PipedInputStream;
 import java.io.PipedOutputStream;
 import java.lang.reflect.InvocationTargetException;
-            cjones
+import java.math.BigInteger;
-            tao
+import java.sql.SQLException;
-            leinfelder
+import java.util.ArrayList;
-            cjones
+import java.util.Calendar;
-            cjones
+import java.util.Date;
-            leinfelder
+import java.util.List;
-            cjones
+import java.util.concurrent.locks.Lock;
             cjones
-            leinfelder
+import javax.servlet.http.HttpServletRequest;
-            tao
+import org.apache.commons.beanutils.BeanUtils;
-            cjones
+import org.apache.log4j.Logger;
-            leinfelder
+import org.dataone.client.v2.CNode;
-            tao
+import org.dataone.client.v2.MNode;
-            leinfelder
+import org.dataone.client.v2.itk.D1Client;
 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;
-            tao
+import org.dataone.service.cn.v2.CNView;
-            cjones
+import org.dataone.service.exceptions.BaseException;
-            cjones
+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;
-            cjones
+import org.dataone.service.exceptions.UnsupportedType;
-            cjones
+import org.dataone.service.exceptions.VersionMismatch;
-            cjones
+import org.dataone.service.types.v1.AccessPolicy;
-            leinfelder
+import org.dataone.service.types.v1.Checksum;
-            leinfelder
+import org.dataone.service.types.v1.ChecksumAlgorithmList;
-            leinfelder
+import org.dataone.service.types.v1.Event;
-            leinfelder
+import org.dataone.service.types.v1.Identifier;
-            cjones
+import org.dataone.service.types.v1.NodeReference;
-            cjones
+import org.dataone.service.types.v1.NodeType;
-            leinfelder
+import org.dataone.service.types.v1.ObjectFormatIdentifier;
 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;
-            tao
+import org.dataone.service.types.v1_1.QueryEngineDescription;
 import org.dataone.service.types.v1_1.QueryEngineList;
 import org.dataone.service.types.v2.Node;
 import org.dataone.service.types.v2.NodeList;
 import org.dataone.service.types.v2.ObjectFormat;
 import org.dataone.service.types.v2.ObjectFormatList;
-            leinfelder
+import org.dataone.service.types.v2.SystemMetadata;
 import org.dataone.service.types.v2.util.ServiceMethodRestrictionUtil;
-            tao
+import org.dataone.service.util.TypeMarshaller;
 import org.jibx.runtime.JiBXException;
             cjones
-            tao
+import edu.ucsb.nceas.metacat.DBUtil;
-            leinfelder
+import edu.ucsb.nceas.metacat.EventLog;
 import edu.ucsb.nceas.metacat.IdentifierManager;
-            cjones
+import edu.ucsb.nceas.metacat.McdbDocNotFoundException;
-            leinfelder
+import edu.ucsb.nceas.metacat.dataone.hazelcast.HazelcastService;
-            tao
+import edu.ucsb.nceas.metacat.properties.PropertyService;
 import edu.ucsb.nceas.utilities.PropertyNotFoundException;
             leinfelder
-            cjones
+/**
  * Represents Metacat's implementation of the DataONE Coordinating Node
-            cjones
+ * service API. Methods implement the various CN* interfaces, and methods common
-            cjones
+ * to both Member Node and Coordinating Node interfaces are found in the
  * D1NodeService super class.
+ *
  */
 public class CNodeService extends D1NodeService implements CNAuthorization,
-            tao
+    CNCore, CNRead, CNReplication, CNView {
             cjones
-            cjones
+  /* the logger instance */
   private Logger logMetacat = null;
-            tao
+  public final static String V2V1MISSMATCH = "The Coordinating Node is not authorized to make systemMetadata changes on this object. Please make changes directly on the authoritative Member Node.";
             cjones
-            cjones
+  /**
    * singleton accessor
    */
-            leinfelder
+  public static CNodeService getInstance(HttpServletRequest request) {
     return new CNodeService(request);
             cjones
   /**
    * Constructor, private for singleton access
    */
-            leinfelder
+  private CNodeService(HttpServletRequest request) {
     super(request);
-            cjones
+    logMetacat = Logger.getLogger(CNodeService.class);
+  }
-            cjones
+  /**
    * Set the replication policy for an object given the object identifier
-            tao
+   * It only is applied to objects whose authoritative mn is a v1 node.
-            cjones
+   * @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
-            cjones
+   * @throws VersionMismatch
             cjones
    */
-            jones
+  @Override
-            cjones
+  public boolean setReplicationPolicy(Session session, Identifier pid,
-            cjones
+      ReplicationPolicy policy, long serialVersion)
-            cjones
+      throws NotImplemented, NotFound, NotAuthorized, ServiceFailure,
-            cjones
+      InvalidRequest, InvalidToken, VersionMismatch {
             cjones
-            tao
+      // do we have a valid pid?
       if (pid == null || pid.getValue().trim().equals("")) {
           throw new InvalidRequest("4883", "The provided identifier was invalid.");
+      }
-            tao
+      //only allow pid to be passed
       String serviceFailure = "4882";
       String notFound = "4884";
       checkV1SystemMetaPidExist(pid, serviceFailure, "The object for given PID "+pid.getValue()+" couldn't be identified if it exists",  notFound,
               "No object could be found for given PID: "+pid.getValue());
-            tao
+      /*String serviceFailureCode = "4882";
-            tao
+      Identifier sid = getPIDForSID(pid, serviceFailureCode);
       if(sid != null) {
           pid = sid;
-            tao
+      }*/
-            cjones
+      // The lock to be used for this identifier
-            cjones
+      Lock lock = null;
             cjones
-            cjones
+      // get the subject
       Subject subject = session.getSubject();
       // are we allowed to do this?
-            cjones
+      if (!isAuthorized(session, pid, Permission.CHANGE_PERMISSION)) {
           throw new NotAuthorized("4881", Permission.CHANGE_PERMISSION
                   + " not allowed by " + subject.getValue() + " on "
                   + pid.getValue());
             cjones
       SystemMetadata systemMetadata = null;
-            cjones
+      try {
-            cjones
+          lock = HazelcastService.getInstance().getLock(pid.getValue());
           lock.lock();
-            cjones
+          logMetacat.debug("Locked identifier " + pid.getValue());
             cjones
           try {
               if ( HazelcastService.getInstance().getSystemMetadataMap().containsKey(pid) ) {
                   systemMetadata = HazelcastService.getInstance().getSystemMetadataMap().get(pid);
+              }
             cjones
               // did we get it correctly?
               if ( systemMetadata == null ) {
                   throw new NotFound("4884", "Couldn't find an object identified by " + pid.getValue());
+              }
-            tao
+              D1NodeVersionChecker checker = new D1NodeVersionChecker(systemMetadata.getAuthoritativeMemberNode());
               String version = checker.getVersion("MNStorage");
               if(version == null) {
-            tao
+                  throw new ServiceFailure("4882", "Couldn't determine the authoritative member node storage version for the pid "+pid.getValue());
-            tao
+              } else if (version.equalsIgnoreCase(D1NodeVersionChecker.V2)) {
                   //we don't apply this method to an object whose authoritative node is v2
                   throw new NotAuthorized("4881", V2V1MISSMATCH);
               } else if (!version.equalsIgnoreCase(D1NodeVersionChecker.V1)) {
                   //we don't understand this version (it is not v1 or v2)
                   throw new InvalidRequest("4883", "The version of the MNStorage is "+version+" for the authoritative member node of the object "+pid.getValue()+". We don't support it.");
+              }
-            cjones
+              // 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.";
-            cjones
+                 throw new VersionMismatch("4886", msg);
             cjones
             cjones
-            cjones
+          } catch (RuntimeException e) { // Catch is generic since HZ throws RuntimeException
-            cjones
+              throw new NotFound("4884", "No record found for: " + pid.getValue());
             cjones
             cjones
           // 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);
-            cjones
+              notifyReplicaNodes(systemMetadata);
-            cjones
+          } catch (RuntimeException e) {
-            cjones
+              throw new ServiceFailure("4882", e.getMessage());
             cjones
-            cjones
+      } catch (RuntimeException e) {
           throw new ServiceFailure("4882", e.getMessage());
       } finally {
           lock.unlock();
           logMetacat.debug("Unlocked identifier " + pid.getValue());
+      }
             cjones
-            cjones
+      return true;
             cjones
             cjones
-            cjones
+  /**
-            leinfelder
+   * 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
-            leinfelder
+   * @throws VersionMismatch
-            leinfelder
+   */
-            leinfelder
+  @Override
-            leinfelder
+  public boolean deleteReplicationMetadata(Session session, Identifier pid, NodeReference nodeId, long serialVersion)
-            leinfelder
+  	throws InvalidToken, ServiceFailure, NotAuthorized, NotFound, NotImplemented, VersionMismatch {
             leinfelder
-            leinfelder
+	  	// The lock to be used for this identifier
 		Lock lock = null;
 		// get the subject
 		Subject subject = session.getSubject();
 		// are we allowed to do this?
-            tao
+		/*boolean isAuthorized = false;
-            cjones
+		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());
             leinfelder
-            tao
+		}*/
 		if(session == null) {
 		    throw new NotAuthorized("4882", "Session cannot be null. It is not authorized for deleting the replication metadata of the object "+pid.getValue());
 		} else {
 		    if(!isCNAdmin(session)) {
 		        throw new NotAuthorized("4882", "The client -"+ session.getSubject().getValue()+ "is not a CN and is not authorized for deleting the replication metadata of the object "+pid.getValue());
+		    }
             leinfelder
 		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?
-            tao
+			/*List<Node> nodeList = D1Client.getCN().listNodes().getNodeList();
-            leinfelder
+			boolean isAllowed = ServiceMethodRestrictionUtil.isMethodAllowed(session.getSubject(), nodeList, "CNReplication", "deleteReplicationMetadata");
-            cjones
+			if (!isAllowed) {
-            leinfelder
+				throw new NotAuthorized("4881", "Caller is not authorized to deleteReplicationMetadata");
-            tao
+			}*/
             leinfelder
 			// delete the replica from the given node
-            cjones
+			// CSJ: use CN.delete() to truly delete a replica, semantically
 			// deleteReplicaMetadata() only modifies the sytem metadata entry.
-            cjones
+			//D1Client.getMN(nodeId).delete(session, pid);
             leinfelder
 			// 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));
-            tao
+				//we don't need to update the modification date.
 				//systemMetadata.setDateSysMetadataModified(Calendar.getInstance().getTime());
-            leinfelder
+				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;
             leinfelder
+  }
-            leinfelder
+  /**
-            cjones
+   * Deletes an object from the Coordinating Node
             leinfelder
    * @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 {
             cjones
-            cjones
+      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
             leinfelder
-            cjones
+      // 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.");
+      }
             tao
       String serviceFailureCode = "4962";
       Identifier sid = getPIDForSID(pid, serviceFailureCode);
       if(sid != null) {
           pid = sid;
+      }
             cjones
-            leinfelder
+	  // check that it is CN/admin
-            leinfelder
+	  boolean allowed = isAdminAuthorized(session);
             leinfelder
-            tao
+	  // additional check if it is the authoritative node if it is not the admin
       if(!allowed) {
           allowed = isAuthoritativeMNodeAdmin(session, pid);
             cjones
             tao
-            leinfelder
+	  if (!allowed) {
-            cjones
+		  String msg = "The subject " + session.getSubject().getValue() +
 			  " is not allowed to call delete() on a Coordinating Node.";
-            leinfelder
+		  logMetacat.info(msg);
-            cjones
+		  throw new NotAuthorized("4960", msg);
             leinfelder
-            cjones
+	  // Don't defer to superclass implementation without a locally registered identifier
-            tao
+	  SystemMetadata systemMetadata = HazelcastService.getInstance().getSystemMetadataMap().get(pid);
-            cjones
+      // 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 ) {
-            tao
+				/*sysMeta.setSerialVersion(sysMeta.getSerialVersion().add(BigInteger.ONE));
-            cjones
+				sysMeta.setArchived(true);
 				sysMeta.setDateSysMetadataModified(Calendar.getInstance().getTime());
-            tao
+				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);
-            tao
+	            String username = session.getSubject().getValue();//just for logging purpose
                 //since data objects were not registered in the identifier table, we use pid as the docid
                 EventLog.getInstance().log(request.getRemoteAddr(), request.getHeader("User-Agent"), username, pid.getValue(), Event.DELETE.xmlValue());
             cjones
 			  } 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());
+		  }
-            leinfelder
+          // NOTE: cannot log the delete without localId
 //          EventLog.getInstance().log(request.getRemoteAddr(),
 //                  request.getHeader("User-Agent"), session.getSubject().getValue(),
 //                  pid.getValue(), Event.DELETE.xmlValue());
             cjones
-            tao
+      } catch (SQLException e) {
           throw new ServiceFailure("4962", "Couldn't delete " + pid.getValue() +
                   ". The local id of the object with the identifier can't be identified since " + e.getMessage());
             cjones
-            cjones
+      // 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());
+      }
-            leinfelder
+	  // notify the replicas
 	  if (systemMetadata.getReplicaList() != null) {
 		  for (Replica replica: systemMetadata.getReplicaList()) {
 			  NodeReference replicaNode = replica.getReplicaMemberNode();
 			  try {
-            cjones
+                  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);
+                  }
-            leinfelder
+			  } catch (Exception e) {
 				  // all we can really do is log errors and carry on with life
-            cjones
+				  logMetacat.error("Error deleting pid: " +  pid.getValue() +
 					  " from replica MN: " + replicaNode.getValue(), e);
             leinfelder
+		  }
+	  }
-            cjones
+	  return pid;
             leinfelder
             leinfelder
   /**
-            cjones
+   * Archives an object from the Coordinating Node
             leinfelder
    * @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 {
-            cjones
+      String localId = null; // The corresponding docid for this pid
 	  Lock lock = null;      // The lock to be used for this identifier
-            cjones
+      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
             cjones
       // 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.");
+      }
-            leinfelder
+	  // check that it is CN/admin
 	  boolean allowed = isAdminAuthorized(session);
-            tao
+	  String serviceFailureCode = "4972";
 	  Identifier sid = getPIDForSID(pid, serviceFailureCode);
 	  if(sid != null) {
 	        pid = sid;
+	  }
-            tao
+	  //check if it is the authoritative member node
 	  if(!allowed) {
 	      allowed = isAuthoritativeMNodeAdmin(session, pid);
+	  }
-            leinfelder
+	  if (!allowed) {
-            cjones
+		  String msg = "The subject " + session.getSubject().getValue() +
 				  " is not allowed to call archive() on a Coordinating Node.";
-            leinfelder
+		  logMetacat.info(msg);
-            cjones
+		  throw new NotAuthorized("4970", msg);
             leinfelder
-            cjones
+      try {
-            tao
+          HazelcastService.getInstance().getSystemMetadataMap().lock(pid);
           logMetacat.debug("CNodeService.archive - lock the system metadata for "+pid.getValue());
-            tao
+          SystemMetadata sysMeta = HazelcastService.getInstance().getSystemMetadataMap().get(pid);
-            tao
+          D1NodeVersionChecker checker = new D1NodeVersionChecker(sysMeta.getAuthoritativeMemberNode());
           String version = checker.getVersion("MNStorage");
           if(version == null) {
-            tao
+              throw new ServiceFailure("4972", "Couldn't determine the authoritative member node storage version for the pid "+pid.getValue());
-            tao
+          } else if (version.equalsIgnoreCase(D1NodeVersionChecker.V2)) {
               //we don't apply this method to an object whose authoritative node is v2
               throw new NotAuthorized("4970", V2V1MISSMATCH);
           } else if (!version.equalsIgnoreCase(D1NodeVersionChecker.V1)) {
               //we don't understand this version (it is not v1 or v2)
               throw new NotImplemented("4974", "The version of the MNStorage is "+version+" for the authoritative member node of the object "+pid.getValue()+". We don't support it.");
+          }
-            tao
+          boolean needModifyDate = true;
           archiveCNObjectWithNotificationReplica(session, pid, sysMeta, needModifyDate);
             tao
-            tao
+      } finally {
           HazelcastService.getInstance().getSystemMetadataMap().unlock(pid);
           logMetacat.debug("CNodeService.archive - unlock the system metadata for "+pid.getValue());
             cjones
             cjones
-            cjones
+	  return pid;
             leinfelder
+  }
             tao
-            leinfelder
+  /**
-            tao
+   * Archive a object on cn and notify the replica. This method doesn't lock the system metadata map. The caller should lock it.
-            tao
+   * This method doesn't check the authorization; this method only accept a pid.
    * @param session
    * @param pid
    * @param sysMeta
    * @param notifyReplica
    * @return
    * @throws InvalidToken
    * @throws ServiceFailure
    * @throws NotAuthorized
    * @throws NotFound
    * @throws NotImplemented
    */
-            tao
+  private Identifier archiveCNObjectWithNotificationReplica(Session session, Identifier pid, SystemMetadata sysMeta, boolean needModifyDate)
                   throws InvalidToken, ServiceFailure, NotAuthorized, NotFound, NotImplemented {
-            tao
+          boolean logArchive = true;
           archiveCNObject(logArchive, session, pid, sysMeta, needModifyDate);
-            tao
+          // notify the replicas
           notifyReplicaNodes(sysMeta);
-            tao
+          return pid;
+    }
   /**
-            leinfelder
+   * 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
    */
-            leinfelder
+  @Override
-            leinfelder
+  public boolean setObsoletedBy(Session session, Identifier pid,
 			Identifier obsoletedByPid, long serialVersion)
 			throws NotImplemented, NotFound, NotAuthorized, ServiceFailure,
 			InvalidRequest, InvalidToken, VersionMismatch {
             tao
       // do we have a valid pid?
       if (pid == null || pid.getValue().trim().equals("")) {
           throw new InvalidRequest("4942", "The provided identifier was invalid.");
+      }
-            tao
+      /*String serviceFailureCode = "4941";
-            tao
+      Identifier sid = getPIDForSID(pid, serviceFailureCode);
       if(sid != null) {
           pid = sid;
-            tao
+      }*/
             tao
       // do we have a valid pid?
       if (obsoletedByPid == null || obsoletedByPid.getValue().trim().equals("")) {
           throw new InvalidRequest("4942", "The provided obsoletedByPid was invalid.");
+      }
       try {
           if(IdentifierManager.getInstance().systemMetadataSIDExists(obsoletedByPid)) {
               throw new InvalidRequest("4942", "The provided obsoletedByPid "+obsoletedByPid.getValue()+" is an existing SID. However, it must NOT be an SID.");
+          }
       } catch (SQLException ee) {
           throw new ServiceFailure("4941", "Couldn't determine if the obsoletedByPid "+obsoletedByPid.getValue()+" is an SID or not. The id shouldn't be an SID.");
+      }
             leinfelder
 		// The lock to be used for this identifier
 		Lock lock = null;
 		// get the subject
 		Subject subject = session.getSubject();
 		// are we allowed to do this?
-            cjones
+		if (!isAuthorized(session, pid, Permission.WRITE)) {
 			throw new NotAuthorized("4881", Permission.WRITE
 					+ " not allowed by " + subject.getValue() + " on "
 					+ pid.getValue());
             leinfelder
+		}
             cjones
-            leinfelder
+		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);
+				}
             tao
 				 //only apply to the object whose authoritative member node is v1.
 	              D1NodeVersionChecker checker = new D1NodeVersionChecker(systemMetadata.getAuthoritativeMemberNode());
 	              String version = checker.getVersion("MNStorage");
 	              if(version == null) {
-            tao
+	                  throw new ServiceFailure("4941", "Couldn't determine the authoritative member node storage version for the pid "+pid.getValue());
-            tao
+	              } else if (version.equalsIgnoreCase(D1NodeVersionChecker.V2)) {
 	                  //we don't apply this method to an object whose authoritative node is v2
 	                  throw new NotAuthorized("4945", V2V1MISSMATCH);
 	              } else if (!version.equalsIgnoreCase(D1NodeVersionChecker.V1)) {
 	                  //we don't understand this version (it is not v1 or v2)
 	                  throw new InvalidRequest("4942", "The version of the MNStorage is "+version+" for the authoritative member node of the object "+pid.getValue()+". We don't support it.");
+	              }
             leinfelder
 			} 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;
+	}
             leinfelder
             leinfelder
-            leinfelder
+  /**
-            cjones
+   * 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
+   *
    */
-            jones
+  @Override
-            cjones
+  public boolean setReplicationStatus(Session session, Identifier pid,
-            cjones
+      NodeReference targetNode, ReplicationStatus status, BaseException failure)
-            cjones
+      throws ServiceFailure, NotImplemented, InvalidToken, NotAuthorized,
       InvalidRequest, NotFound {
             leinfelder
 	  // cannot be called by public
 	  if (session == null) {
 		  throw new NotAuthorized("4720", "Session cannot be null");
             tao
 	  /*else {
-            tao
+	      if(!isCNAdmin(session)) {
               throw new NotAuthorized("4720", "The client -"+ session.getSubject().getValue()+ "is not a CN and is not authorized for setting the replication status of the object "+pid.getValue());
+        }
-            tao
+	  }*/
             tao
 	// do we have a valid pid?
       if (pid == null || pid.getValue().trim().equals("")) {
           throw new InvalidRequest("4730", "The provided identifier was invalid.");
+      }
             cjones
-            tao
+      /*String serviceFailureCode = "4700";
-            tao
+      Identifier sid = getPIDForSID(pid, serviceFailureCode);
       if(sid != null) {
           pid = sid;
-            tao
+      }*/
             tao
-            cjones
+      // The lock to be used for this identifier
-            cjones
+      Lock lock = null;
             cjones
-            cjones
+      boolean allowed = false;
       int replicaEntryIndex = -1;
       List<Replica> replicas = null;
       // get the subject
       Subject subject = session.getSubject();
-            cjones
+      logMetacat.debug("ReplicationStatus for identifier " + pid.getValue() +
           " is " + status.toString());
             cjones
       SystemMetadata systemMetadata = null;
             cjones
       try {
-            cjones
+          lock = HazelcastService.getInstance().getLock(pid.getValue());
-            cjones
+          lock.lock();
-            cjones
+          logMetacat.debug("Locked identifier " + pid.getValue());
-            cjones
+          try {
               systemMetadata = HazelcastService.getInstance().getSystemMetadataMap().get(pid);
             cjones
-            cjones
+              // did we get it correctly?
-            cjones
+              if ( systemMetadata == null ) {
                   logMetacat.debug("systemMetadata is null for " + pid.getValue());
-            cjones
+                  throw new NotFound("4740", "Couldn't find an object identified by " + pid.getValue());
             cjones
+              }
               replicas = systemMetadata.getReplicaList();
               int count = 0;
             cjones
-            cjones
+              // was there a failure? log it
-            cjones
+              if ( failure != null && status.equals(ReplicationStatus.FAILED) ) {
-            cjones
+                 String msg = "The replication request of the object identified by " +
                      pid.getValue() + " failed.  The error message was " +
                      failure.getMessage() + ".";
-            tao
+                 logMetacat.error(msg);
             cjones
             cjones
-            cjones
+              if (replicas.size() > 0 && replicas != null) {
                   // find the target replica index in the replica list
                   for (Replica replica : replicas) {
-            cjones
+                      String replicaNodeStr = replica.getReplicaMemberNode().getValue();
-            cjones
+                      String targetNodeStr = targetNode.getValue();
-            cjones
+                      logMetacat.debug("Comparing " + replicaNodeStr + " to " + targetNodeStr);
             cjones
-            cjones
+                      if (replicaNodeStr.equals(targetNodeStr)) {
                           replicaEntryIndex = count;
                           logMetacat.debug("replica entry index is: "
                                   + replicaEntryIndex);
                           break;
+                      }
                       count++;
             cjones
+                  }
+              }
               // are we allowed to do this? only CNs and target MNs are allowed
               CNode cn = D1Client.getCN();
               List<Node> nodes = cn.listNodes().getNodeList();
             cjones
-            cjones
+              // find the node in the node list
               for ( Node node : nodes ) {
                   NodeReference nodeReference = node.getIdentifier();
-            cjones
+                  logMetacat.debug("In setReplicationStatus(), Node reference is: " +
                       nodeReference.getValue());
             cjones
-            cjones
+                  // allow target MN certs
-            cjones
+                  if ( targetNode.getValue().equals(nodeReference.getValue() ) &&
                       node.getType().equals(NodeType.MN)) {
-            cjones
+                      List<Subject> nodeSubjects = node.getSubjectList();
                       // check if the session subject is in the node subject list
                       for (Subject nodeSubject : nodeSubjects) {
-            cjones
+                          logMetacat.debug("In setReplicationStatus(), comparing subjects: " +
                                   nodeSubject.getValue() + " and " + subject.getValue());
-            cjones
+                          if ( nodeSubject.equals(subject) ) { // subject of session == target node subject
             cjones
-            cjones
+                              // lastly limit to COMPLETED, INVALIDATED,
                               // and FAILED status updates from MNs only
-            cjones
+                              if ( status.equals(ReplicationStatus.COMPLETED) ||
                                    status.equals(ReplicationStatus.INVALIDATED) ||
                                    status.equals(ReplicationStatus.FAILED)) {
-            cjones
+                                  allowed = true;
                                   break;
+                              }
             cjones
+                      }
+                  }
+              }
             cjones
-            cjones
+              if ( !allowed ) {
                   //check for CN admin access
-            tao
+                  //allowed = isAuthorized(session, pid, Permission.WRITE);
                   allowed = isCNAdmin(session);
             cjones
             cjones
               if ( !allowed ) {
                   String msg = "The subject identified by "
                           + subject.getValue()
-            tao
+                          + " is not a CN or MN, and does not have permission to set the replication status for "
-            cjones
+                          + "the replica identified by "
                           + targetNode.getValue() + ".";
                   logMetacat.info(msg);
                   throw new NotAuthorized("4720", msg);
             cjones
             cjones
-            cjones
+          } catch (RuntimeException e) { // Catch is generic since HZ throws RuntimeException
             throw new NotFound("4740", "No record found for: " + pid.getValue() +
                 " : " + e.getMessage());
             cjones
-            cjones
+          Replica targetReplica = new Replica();
-            cjones
+          // set the status for the replica
           if ( replicaEntryIndex != -1 ) {
-            cjones
+              targetReplica = replicas.get(replicaEntryIndex);
             cjones
               // don't allow status to change from COMPLETED to anything other
               // than INVALIDATED: prevents overwrites from race conditions
-            cjones
+              if ( targetReplica.getReplicationStatus().equals(ReplicationStatus.COMPLETED) &&
             	   !status.equals(ReplicationStatus.INVALIDATED)) {
-            cjones
+            	  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());
+              }
-            tao
+              if(targetReplica.getReplicationStatus().equals(status)) {
                   //There is no change in the status, we do nothing.
                   return true;
+              }
-            cjones
+              targetReplica.setReplicationStatus(status);
             cjones
-            cjones
+              logMetacat.debug("Set the replication status for " +
                   targetReplica.getReplicaMemberNode().getValue() + " to " +
-            cjones
+                  targetReplica.getReplicationStatus() + " for identifier " +
                   pid.getValue());
             cjones
-            cjones
+          } else {
-            cjones
+              // this is a new entry, create it
               targetReplica.setReplicaMemberNode(targetNode);
               targetReplica.setReplicationStatus(status);
               targetReplica.setReplicaVerified(Calendar.getInstance().getTime());
               replicas.add(targetReplica);
             cjones
-            cjones
+          systemMetadata.setReplicaList(replicas);
           // update the metadata
           try {
               systemMetadata.setSerialVersion(systemMetadata.getSerialVersion().add(BigInteger.ONE));
-            cjones
+              // Based on CN behavior discussion 9/16/15, we no longer want to
               // update the modified date for changes to the replica list
               //systemMetadata.setDateSysMetadataModified(Calendar.getInstance().getTime());
-            cjones
+              HazelcastService.getInstance().getSystemMetadataMap().put(systemMetadata.getIdentifier(), systemMetadata);
             cjones
-            cjones
+              if ( !status.equals(ReplicationStatus.QUEUED) &&
             	   !status.equals(ReplicationStatus.REQUESTED)) {
             cjones
                 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());
+              }
-            cjones
+              if ( status.equals(ReplicationStatus.FAILED) && failure != null ) {
-            cjones
+                  logMetacat.warn("Replication failed for identifier " + pid.getValue() +
                       " on target node " + targetNode + ". The exception was: " +
                       failure.getMessage());
+              }
             cjones
-            tao
+			  // update the replica nodes about the completed replica when complete, failed or invalid
               if (status.equals(ReplicationStatus.COMPLETED) || status.equals(ReplicationStatus.FAILED) ||
                       status.equals(ReplicationStatus.INVALIDATED)) {
-            cjones
+				notifyReplicaNodes(systemMetadata);
             cjones
-            cjones
+          } catch (RuntimeException e) {
-            cjones
+              throw new ServiceFailure("4700", e.getMessage());
             cjones
-            cjones
+    } catch (RuntimeException e) {
         String msg = "There was a RuntimeException getting the lock for " +
             pid.getValue();
         logMetacat.info(msg);
     } finally {
-            cjones
+        lock.unlock();
         logMetacat.debug("Unlocked identifier " + pid.getValue());
             cjones
             cjones
             cjones
-            cjones
+      return true;
             cjones
-            cjones
+/**
-            cjones
+   * 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
    */
-            jones
+  @Override
-            cjones
+  public Checksum getChecksum(Session session, Identifier pid)
     throws InvalidToken, ServiceFailure, NotAuthorized, NotFound,
-            leinfelder
+    NotImplemented {
             leinfelder
 	boolean isAuthorized = false;
 	try {
 		isAuthorized = isAuthorized(session, pid, Permission.READ);
 	} catch (InvalidRequest e) {
 		throw new ServiceFailure("1410", e.getDescription());
+	}
     if (!isAuthorized) {
-            cjones
+        throw new NotAuthorized("1400", Permission.READ + " not allowed on " + pid.getValue());
             cjones
             cjones
-            cjones
+    SystemMetadata systemMetadata = null;
-            cjones
+    Checksum checksum = null;
-            cjones
+    try {
-            cjones
+        systemMetadata = HazelcastService.getInstance().getSystemMetadataMap().get(pid);
         if (systemMetadata == null ) {
-            tao
+            String error ="";
             String localId = null;
             try {
                 localId = IdentifierManager.getInstance().getLocalId(pid.getValue());
              } catch (Exception e) {
                 logMetacat.warn("Couldn't find the local id for the pid "+pid.getValue());
+            }
             if(localId != null && EventLog.getInstance().isDeleted(localId)) {
                 error = DELETEDMESSAGE;
-            tao
+            } else if (localId == null && EventLog.getInstance().isDeleted(pid.getValue())) {
                 error = DELETEDMESSAGE;
             tao
             throw new NotFound("1420", "Couldn't find an object identified by " + pid.getValue()+". "+error);
             cjones
-            cjones
+        checksum = systemMetadata.getChecksum();
             cjones
     } catch (RuntimeException e) {
-            cjones
+        throw new ServiceFailure("1410", "An error occurred getting the checksum for " +
             pid.getValue() + ". The error message was: " + e.getMessage());
             cjones
     return checksum;
+  }
             cjones
-            cjones
+  /**
    * Resolve the location of a given object
+   *
    * @param session - the Session object containing the credentials for the Subject
    * @param pid - the object identifier for the given object
+   *
    * @return objectLocationList - the list of nodes known to contain the object
+   *
    * @throws InvalidToken
    * @throws ServiceFailure
    * @throws NotAuthorized
    * @throws NotFound
    * @throws NotImplemented
    */
-            jones
+  @Override
-            cjones
+  public ObjectLocationList resolve(Session session, Identifier pid)
-            leinfelder
+    throws InvalidToken, ServiceFailure, NotAuthorized,
-            cjones
+    NotFound, NotImplemented {
             cjones
-            cjones
+    throw new NotImplemented("4131", "resolve not implemented");
             leinfelder
             cjones
             cjones
-            cjones
+  /**
-            leinfelder
+   * Metacat does not implement this method at the CN level
-            cjones
+   */
-            jones
+  @Override
-            cjones
+  public ObjectList search(Session session, String queryType, String query)
     throws InvalidToken, ServiceFailure, NotAuthorized, InvalidRequest,
     NotImplemented {
             cjones
-            leinfelder
+		  throw new NotImplemented("4281", "Metacat does not implement CN.search");
 //    ObjectList objectList = null;
 //    try {
 //        objectList =
 //          IdentifierManager.getInstance().querySystemMetadata(
 //              null, //startTime,
 //              null, //endTime,
 //              null, //objectFormat,
 //              false, //replicaStatus,
 //              0, //start,
 //              1000 //count
 //              );
 //
 //    } catch (Exception e) {
 //      throw new ServiceFailure("4310", "Error querying system metadata: " + e.getMessage());
 //    }
 //
 //      return objectList;
             cjones
   /**
    * Returns the object format registered in the DataONE Object Format
    * Vocabulary for the given format identifier
+   *
    * @param fmtid - the identifier of the format requested
+   *
    * @return objectFormat - the object format requested
+   *
    * @throws ServiceFailure
    * @throws NotFound
    * @throws InsufficientResources
    * @throws NotImplemented
    */
-            jones
+  @Override
-            cjones
+  public ObjectFormat getFormat(ObjectFormatIdentifier fmtid)
-            leinfelder
+    throws ServiceFailure, NotFound, NotImplemented {
             cjones
       return ObjectFormatService.getInstance().getFormat(fmtid);
+  }
             cjones
-            tao
+    @Override
     public ObjectFormatIdentifier addFormat(Session session, ObjectFormatIdentifier formatId, ObjectFormat format)
             throws ServiceFailure, NotFound, NotImplemented, NotAuthorized, InvalidToken {
         logMetacat.debug("CNodeService.addFormat() called.\n" +
                 "format ID: " + format.getFormatId() + "\n" +
                 "format name: " + format.getFormatName() + "\n" +
                 "format type: " + format.getFormatType() );
         // FIXME remove:
         if (true)
             throw new NotImplemented("0000", "Implementation underway... Will need testing too...");
         if (!isAdminAuthorized(session))
             throw new NotAuthorized("0000", "Not authorized to call addFormat()");
         String separator = ".";
         try {
             separator = PropertyService.getProperty("document.accNumSeparator");
         } catch (PropertyNotFoundException e) {
             logMetacat.warn("Unable to find property \"document.accNumSeparator\"\n" + e.getMessage());
+        }
         // find pids of last and next ObjectFormatList
         String OBJECT_FORMAT_DOCID = ObjectFormatService.OBJECT_FORMAT_DOCID;
         int lastRev = -1;
         try {
             lastRev = DBUtil.getLatestRevisionInDocumentTable(OBJECT_FORMAT_DOCID);
         } catch (SQLException e) {
             throw new ServiceFailure("0000", "Unable to locate last revision of the object format list.\n" + e.getMessage());
+        }
         int nextRev = lastRev + 1;
         String lastDocID = OBJECT_FORMAT_DOCID + separator + lastRev;
         String nextDocID = OBJECT_FORMAT_DOCID + separator + nextRev;
         Identifier lastPid = new Identifier();
         lastPid.setValue(lastDocID);
         Identifier nextPid = new Identifier();
         nextPid.setValue(nextDocID);
         logMetacat.debug("Last ObjectFormatList document ID: " + lastDocID + "\n"
                 + "Next ObjectFormatList document ID: " + nextDocID);
         // add new format to the current ObjectFormatList
         ObjectFormatList objectFormatList = ObjectFormatService.getInstance().listFormats();
         List<ObjectFormat> innerList = objectFormatList.getObjectFormatList();
         innerList.add(format);
         // get existing (last) sysmeta and make a copy
         SystemMetadata lastSysmeta = getSystemMetadata(session, lastPid);
         SystemMetadata nextSysmeta = new SystemMetadata();
         try {
             BeanUtils.copyProperties(nextSysmeta, lastSysmeta);
         } catch (IllegalAccessException | InvocationTargetException e) {
             throw new ServiceFailure("0000", "Unable to create system metadata for updated object format list.\n" + e.getMessage());
+        }
         // create the new object format list, and update the old sysmeta with obsoletedBy
         createNewObjectFormatList(session, lastPid, nextPid, objectFormatList, nextSysmeta);
         updateOldObjectFormatList(session, lastPid, nextPid, lastSysmeta);
         // TODO add to ObjectFormatService local cache?
         return formatId;
+    }
     /**
      * Creates the object for the next / updated version of the ObjectFormatList.
+     *
      * @param session
      * @param lastPid
      * @param nextPid
      * @param objectFormatList
      * @param lastSysmeta
      */
     private void createNewObjectFormatList(Session session, Identifier lastPid, Identifier nextPid,
             ObjectFormatList objectFormatList, SystemMetadata lastSysmeta)
                     throws InvalidToken, ServiceFailure, NotAuthorized, NotImplemented {
         PipedInputStream is = new PipedInputStream();
         PipedOutputStream os = null;
         try {
             os = new PipedOutputStream(is);
             TypeMarshaller.marshalTypeToOutputStream(objectFormatList, os);
         } catch (JiBXException | IOException e) {
             throw new ServiceFailure("0000", "Unable to marshal object format list.\n" + e.getMessage());
         } finally {
             try {
                 os.flush();
                 os.close();
             } catch (IOException ioe) {
                 throw new ServiceFailure("0000", "Unable to marshal object format list.\n" + ioe.getMessage());
+            }
+        }
         BigInteger docSize = lastSysmeta.getSize();
         try {
             docSize = BigInteger.valueOf(is.available());
         } catch (IOException e) {
             logMetacat.warn("Unable to set an accurate size for the new object format list.", e);
+        }
         lastSysmeta.setIdentifier(nextPid);
         lastSysmeta.setObsoletes(lastPid);
         lastSysmeta.setSize(docSize);
         lastSysmeta.setSubmitter(session.getSubject());
         lastSysmeta.setDateUploaded(new Date());
         // create new object format list
         try {
             create(session, nextPid, is, lastSysmeta);
         } catch (IdentifierNotUnique | UnsupportedType | InsufficientResources
                 | InvalidSystemMetadata | InvalidRequest e) {
             throw new ServiceFailure("0000", "Unable to create() new object format list" + e.getMessage());
+        }
+    }
     /**
      * Updates the SystemMetadata for the old version of the ObjectFormatList
      * by setting the obsoletedBy value to the pid of the new version of the
      * ObjectFormatList.
+     *
      * @param session
      * @param lastPid
      * @param obsoletedByPid
      * @param lastSysmeta
      * @throws ServiceFailure
      */
     private void updateOldObjectFormatList(Session session, Identifier lastPid, Identifier obsoletedByPid, SystemMetadata lastSysmeta)
             throws ServiceFailure {
         lastSysmeta.setObsoletedBy(obsoletedByPid);
         try {
             this.updateSystemMetadata(session, lastPid, lastSysmeta);
         } catch (NotImplemented | NotAuthorized | ServiceFailure | InvalidRequest
                 | InvalidSystemMetadata | InvalidToken e) {
             throw new ServiceFailure("0000", "Unable to update metadata of old object format list.\n" + e.getMessage());
+        }
+    }
-            cjones
+  /**
-            cjones
+   * Returns a list of all object formats registered in the DataONE Object
    * Format Vocabulary
             cjones
    * @return objectFormatList - The list of object formats registered in
    *                            the DataONE Object Format Vocabulary
+   *
    * @throws ServiceFailure
    * @throws NotImplemented
    * @throws InsufficientResources
    */
-            jones
+  @Override
-            cjones
+  public ObjectFormatList listFormats()
-            leinfelder
+    throws ServiceFailure, NotImplemented {
             cjones
-            cjones
+    return ObjectFormatService.getInstance().listFormats();
+  }
             cjones
-            cjones
+  /**
-            cjones
+   * Returns a list of nodes that have been registered with the DataONE infrastructure
             cjones
    * @return nodeList - List of nodes from the registry
+   *
    * @throws ServiceFailure
    * @throws NotImplemented
    */
-            jones
+  @Override
-            cjones
+  public NodeList listNodes()
     throws NotImplemented, ServiceFailure {
             cjones
-            cjones
+    throw new NotImplemented("4800", "listNodes not implemented");
+  }
             cjones
-            cjones
+  /**
-            cjones
+   * Provides a mechanism for adding system metadata independently of its
    * associated object, such as when adding system metadata for data objects.
             cjones
    * @param session - the Session object containing the credentials for the Subject
    * @param pid - The identifier of the object to register the system metadata against
    * @param sysmeta - The system metadata to be registered
+   *
    * @return true if the registration succeeds
+   *
    * @throws NotImplemented
    * @throws NotAuthorized
    * @throws ServiceFailure
    * @throws InvalidRequest
    * @throws InvalidSystemMetadata
    */
-            jones
+  @Override
-            cjones
+  public Identifier registerSystemMetadata(Session session, Identifier pid,
       SystemMetadata sysmeta)
       throws NotImplemented, NotAuthorized, ServiceFailure, InvalidRequest,
       InvalidSystemMetadata {
             cjones
-            cjones
+      // The lock to be used for this identifier
-            cjones
+      Lock lock = null;
             cjones
-            cjones
+      // TODO: control who can call this?
       if (session == null) {
           //TODO: many of the thrown exceptions do not use the correct error codes
           //check these against the docs and correct them
           throw new NotAuthorized("4861", "No Session - could not authorize for registration." +
                   "  If you are not logged in, please do so and retry the request.");
-            tao
+      } else {
           //only CN is allwoed
           if(!isCNAdmin(session)) {
                 throw new NotAuthorized("4861", "The client -"+ session.getSubject().getValue()+ "is not a CN and is not authorized for registering the system metadata of the object "+pid.getValue());
+          }
             cjones
-            tao
+      // the identifier can't be an SID
       try {
           if(IdentifierManager.getInstance().systemMetadataSIDExists(pid)) {
               throw new InvalidRequest("4863", "The provided identifier "+pid.getValue()+" is a series id which is not allowed.");
+          }
       } catch (SQLException sqle) {
           throw new ServiceFailure("4862", "Couldn't determine if the pid "+pid.getValue()+" is a series id since "+sqle.getMessage());
+      }
             cjones
       // verify that guid == SystemMetadata.getIdentifier()
       logMetacat.debug("Comparing guid|sysmeta_guid: " + pid.getValue() +
           "|" + sysmeta.getIdentifier().getValue());
       if (!pid.getValue().equals(sysmeta.getIdentifier().getValue())) {
           throw new InvalidRequest("4863",
               "The identifier in method call (" + pid.getValue() +
               ") does not match identifier in system metadata (" +
               sysmeta.getIdentifier().getValue() + ").");
+      }
             tao
       //check if the sid is legitimate in the system metadata
-            tao
+      //checkSidInModifyingSystemMetadata(sysmeta, "4864", "4862");
       Identifier sid = sysmeta.getSeriesId();
       if(sid != null) {
           if (!isValidIdentifier(sid)) {
               throw new InvalidRequest("4863", "The series id in the system metadata is invalid in the request.");
+          }
+      }
             leinfelder
-            cjones
+      try {
-            cjones
+          lock = HazelcastService.getInstance().getLock(sysmeta.getIdentifier().getValue());
-            cjones
+          lock.lock();
-            cjones
+          logMetacat.debug("Locked identifier " + pid.getValue());
-            cjones
+          logMetacat.debug("Checking if identifier exists...");
           // Check that the identifier does not already exist
           if (HazelcastService.getInstance().getSystemMetadataMap().containsKey(pid)) {
               throw new InvalidRequest("4863",
                   "The identifier is already in use by an existing object.");
             cjones
             cjones
           // insert the system metadata into the object store
           logMetacat.debug("Starting to insert SystemMetadata...");
           try {
-            tao
+              //for the object whose authoriative mn is v1. we need reset the modification date.
               //d1-sync already set the serial version. so we don't need do again.
               D1NodeVersionChecker checker = new D1NodeVersionChecker(sysmeta.getAuthoritativeMemberNode());
               String version = checker.getVersion("MNStorage");
               if(version != null && version.equalsIgnoreCase(D1NodeVersionChecker.V1)) {
                   sysmeta.setDateSysMetadataModified(Calendar.getInstance().getTime());
+              }
-            cjones
+              HazelcastService.getInstance().getSystemMetadataMap().put(sysmeta.getIdentifier(), sysmeta);
-            cjones
+          } catch (RuntimeException e) {
-            cjones
+            logMetacat.error("Problem registering system metadata: " + pid.getValue(), e);
               throw new ServiceFailure("4862", "Error inserting system metadata: " +
                   e.getClass() + ": " + e.getMessage());
+          }
-            cjones
+      } catch (RuntimeException e) {
-            cjones
+          throw new ServiceFailure("4862", "Error inserting system metadata: " +
-            cjones
+                  e.getClass() + ": " + e.getMessage());
             cjones
-            cjones
+      }  finally {
           lock.unlock();
           logMetacat.debug("Unlocked identifier " + pid.getValue());
+      }
             cjones
             cjones
       logMetacat.debug("Returning from registerSystemMetadata");
             leinfelder
       try {
     	  String localId = IdentifierManager.getInstance().getLocalId(pid.getValue());
     	  EventLog.getInstance().log(request.getRemoteAddr(),
     	          request.getHeader("User-Agent"), session.getSubject().getValue(),
     	          localId, "registerSystemMetadata");
       } catch (McdbDocNotFoundException e) {
     	  // do nothing, no localId to log with
     	  logMetacat.warn("Could not log 'registerSystemMetadata' event because no localId was found for pid: " + pid.getValue());
-            tao
+      } catch (SQLException ee) {
           // do nothing, no localId to log with
           logMetacat.warn("Could not log 'registerSystemMetadata' event because the localId couldn't be identified for pid: " + pid.getValue());
             leinfelder
-            cjones
+      return pid;
             cjones
   /**
-            cjones
+   * Given an optional scope and format, reserves and returns an identifier
    * within that scope and format that is unique and will not be
    * used by any other sessions.
             cjones
    * @param session - the Session object containing the credentials for the Subject
    * @param pid - The identifier of the object to register the system metadata against
    * @param scope - An optional string to be used to qualify the scope of
    *                the identifier namespace, which is applied differently
    *                depending on the format requested. If scope is not
    *                supplied, a default scope will be used.
    * @param format - The optional name of the identifier format to be used,
    *                  drawn from a DataONE-specific vocabulary of identifier
    *                 format names, including several common syntaxes such
    *                 as DOI, LSID, UUID, and LSRN, among others. If the
    *                 format is not supplied by the caller, the CN service
    *                 will use a default identifier format, which may change
    *                 over time.
+   *
    * @return true if the registration succeeds
+   *
    * @throws InvalidToken
    * @throws ServiceFailure
    * @throws NotAuthorized
    * @throws IdentifierNotUnique
    * @throws NotImplemented
    */
-            jones
+  @Override
-            leinfelder
+  public Identifier reserveIdentifier(Session session, Identifier pid)
-            cjones
+  throws InvalidToken, ServiceFailure,
-            leinfelder
+        NotAuthorized, IdentifierNotUnique, NotImplemented, InvalidRequest {
             cjones
-            cjones
+    throw new NotImplemented("4191", "reserveIdentifier not implemented on this node");
+  }
-            jones
+  @Override
-            cjones
+  public Identifier generateIdentifier(Session session, String scheme, String fragment)
   throws InvalidToken, ServiceFailure,
-            leinfelder
+        NotAuthorized, NotImplemented, InvalidRequest {
-            cjones
+    throw new NotImplemented("4191", "generateIdentifier not implemented on this node");
+  }
   /**
     * Checks whether the pid is reserved by the subject in the session param
     * If the reservation is held on the pid by the subject, we return true.
+    *
    * @param session - the Session object containing the Subject
    * @param pid - The identifier to check
+   *
    * @return true if the reservation exists for the subject/pid
+   *
    * @throws InvalidToken
    * @throws ServiceFailure
    * @throws NotFound - when the pid is not found (in use or in reservation)
    * @throws NotAuthorized - when the subject does not hold a reservation on the pid
    * @throws IdentifierNotUnique - when the pid is in use
    * @throws NotImplemented
    */
             cjones
-            jones
+  @Override
-            leinfelder
+  public boolean hasReservation(Session session, Subject subject, Identifier pid)
-            rnahf
+      throws InvalidToken, ServiceFailure, NotFound, NotAuthorized,
-            cjones
+      NotImplemented, InvalidRequest {
       throw new NotImplemented("4191", "hasReservation not implemented on this node");
+  }
             leinfelder
-            cjones
+  /**
-            cjones
+   * Changes ownership (RightsHolder) of the specified object to the
    * subject specified by userId
             cjones
    * @param session - the Session object containing the credentials for the Subject
    * @param pid - Identifier of the object to be modified
    * @param userId - The subject that will be taking ownership of the specified object.
+   *
    * @return pid - the identifier of the modified object
+   *
    * @throws ServiceFailure
    * @throws InvalidToken
    * @throws NotFound
    * @throws NotAuthorized
    * @throws NotImplemented
    * @throws InvalidRequest
    */
-            jones
+  @Override
-            leinfelder
+  public Identifier setRightsHolder(Session session, Identifier pid, Subject userId,
-            cjones
+      long serialVersion)
       throws InvalidToken, ServiceFailure, NotFound, NotAuthorized,
-            cjones
+      NotImplemented, InvalidRequest, VersionMismatch {
             cjones
-            cjones
+      // The lock to be used for this identifier
-            cjones
+      Lock lock = null;
             tao
       // do we have a valid pid?
       if (pid == null || pid.getValue().trim().equals("")) {
           throw new InvalidRequest("4442", "The provided identifier was invalid.");
+      }
             cjones
-            cjones
+      // get the subject
       Subject subject = session.getSubject();
-            tao
+      String serviceFailureCode = "4490";
       Identifier sid = getPIDForSID(pid, serviceFailureCode);
       if(sid != null) {
           pid = sid;
+      }
-            cjones
+      // are we allowed to do this?
-            cjones
+      if (!isAuthorized(session, pid, Permission.CHANGE_PERMISSION)) {
           throw new NotAuthorized("4440", "not allowed by "
                   + subject.getValue() + " on " + pid.getValue());
             cjones
       SystemMetadata systemMetadata = null;
-            cjones
+      try {
-            cjones
+          lock = HazelcastService.getInstance().getLock(pid.getValue());
-            leinfelder
+          lock.lock();
-            cjones
+          logMetacat.debug("Locked identifier " + pid.getValue());
-            cjones
+          try {
               systemMetadata = HazelcastService.getInstance().getSystemMetadataMap().get(pid);
-            tao
+              if(systemMetadata == null) {
                   throw new NotFound("4460", "The object "+pid.getValue()+" doesn't exist in the node.");
+              }
-            cjones
+              // 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.";
-            cjones
+                 throw new VersionMismatch("4443", msg);
             cjones
-            tao
+              //only apply to the object whose authoritative member node is v1.
               D1NodeVersionChecker checker = new D1NodeVersionChecker(systemMetadata.getAuthoritativeMemberNode());
               String version = checker.getVersion("MNStorage");
               if(version == null) {
-            tao
+                  throw new ServiceFailure("4490", "Couldn't determine the authoritative member node storage version for the pid "+pid.getValue());
-            tao
+              } else if (version.equalsIgnoreCase(D1NodeVersionChecker.V2)) {
                   //we don't apply this method to an object whose authoritative node is v2
                   throw new NotAuthorized("4440", V2V1MISSMATCH);
               } else if (!version.equalsIgnoreCase(D1NodeVersionChecker.V1)) {
                   //we don't understand this version (it is not v1 or v2)
                   throw new InvalidRequest("4442", "The version of the MNStorage is "+version+" for the authoritative member node of the object "+pid.getValue()+". We don't support it.");
+              }
-            cjones
+          } catch (RuntimeException e) { // Catch is generic since HZ throws RuntimeException
-            cjones
+              throw new NotFound("4460", "No record found for: " + pid.getValue());
             cjones
             cjones
           // set the new rights holder
           systemMetadata.setRightsHolder(userId);
             cjones
-            cjones
+          // update the metadata
           try {
               systemMetadata.setSerialVersion(systemMetadata.getSerialVersion().add(BigInteger.ONE));
               systemMetadata.setDateSysMetadataModified(Calendar.getInstance().getTime());
               HazelcastService.getInstance().getSystemMetadataMap().put(pid, systemMetadata);
-            cjones
+              notifyReplicaNodes(systemMetadata);
             cjones
-            cjones
+          } catch (RuntimeException e) {
               throw new ServiceFailure("4490", e.getMessage());
             cjones
             cjones
             cjones
-            cjones
+      } catch (RuntimeException e) {
-            cjones
+          throw new ServiceFailure("4490", e.getMessage());
             cjones
       } finally {
-            cjones
+          lock.unlock();
-            cjones
+          logMetacat.debug("Unlocked identifier " + pid.getValue());
             cjones
             cjones
-            cjones
+      return pid;
             cjones
             cjones
-            cjones
+  /**
    * Verify that a replication task is authorized by comparing the target node's
    * Subject (from the X.509 certificate-derived Session) with the list of
    * subjects in the known, pending replication tasks map.
+   *
    * @param originatingNodeSession - Session information that contains the
    *                                 identity of the calling user
    * @param targetNodeSubject - Subject identifying the target node
    * @param pid - the identifier of the object to be replicated
    * @param replicatePermission - the execute permission to be granted
+   *
    * @throws ServiceFailure
    * @throws NotImplemented
    * @throws InvalidToken
    * @throws NotAuthorized
    * @throws InvalidRequest
    * @throws NotFound
    */
-            jones
+  @Override
-            cjones
+  public boolean isNodeAuthorized(Session originatingNodeSession,
-            leinfelder
+    Subject targetNodeSubject, Identifier pid)
-            cjones
+    throws NotImplemented, NotAuthorized, InvalidToken, ServiceFailure,
     NotFound, InvalidRequest {
             cjones
-            cjones
+    boolean isAllowed = false;
     SystemMetadata sysmeta = null;
-            cjones
+    NodeReference targetNode = null;
-            cjones
+    try {
       // get the target node reference from the nodes list
       CNode cn = D1Client.getCN();
       List<Node> nodes = cn.listNodes().getNodeList();
             cjones
-            cjones
+      if ( nodes != null ) {
         for (Node node : nodes) {
             cjones
-            leinfelder
+        	if (node.getSubjectList() != null) {
 	            for (Subject nodeSubject : node.getSubjectList()) {
 	                if ( nodeSubject.equals(targetNodeSubject) ) {
 	                    targetNode = node.getIdentifier();
 	                    logMetacat.debug("targetNode is : " + targetNode.getValue());
 	                    break;
+	                }
+	            }
+        	}
             cjones
-            cjones
+            if ( targetNode != null) { break; }
             cjones
       } else {
           String msg = "Couldn't get the node list from the CN";
           logMetacat.debug(msg);
           throw new ServiceFailure("4872", msg);
             cjones
             cjones
       // can't find a node listed with the given subject
       if ( targetNode == null ) {
           String msg = "There is no Member Node registered with a node subject " +
               "matching " + targetNodeSubject.getValue();
           logMetacat.info(msg);
-            leinfelder
+          throw new NotAuthorized("4871", msg);
             cjones
+      }
-            cjones
+      logMetacat.debug("Getting system metadata for identifier " + pid.getValue());
-            cjones
+      sysmeta = HazelcastService.getInstance().getSystemMetadataMap().get(pid);
             cjones
-            cjones
+      if ( sysmeta != null ) {
           List<Replica> replicaList = sysmeta.getReplicaList();
           if ( replicaList != null ) {
               // find the replica with the status set to 'requested'
               for (Replica replica : replicaList) {
                   ReplicationStatus status = replica.getReplicationStatus();
                   NodeReference listedNode = replica.getReplicaMemberNode();
-            cjones
+                  if ( listedNode != null && targetNode != null ) {
                       logMetacat.debug("Comparing " + listedNode.getValue()
                               + " to " + targetNode.getValue());
                       if (listedNode.getValue().equals(targetNode.getValue())
                               && status.equals(ReplicationStatus.REQUESTED)) {
                           isAllowed = true;
                           break;
             cjones
             cjones
             cjones
+              }
             cjones
-            cjones
+          logMetacat.debug("The " + targetNode.getValue() + " is allowed " +
               "to replicate: " + isAllowed + " for " + pid.getValue());
             cjones
       } else {
           logMetacat.debug("System metadata for identifier " + pid.getValue() +
-            tao
+          " is null.");
           String error ="";
           String localId = null;
           try {
               localId = IdentifierManager.getInstance().getLocalId(pid.getValue());
            } catch (Exception e) {
               logMetacat.warn("Couldn't find the local id for the pid "+pid.getValue());
+          }
             cjones
-            tao
+          if(localId != null && EventLog.getInstance().isDeleted(localId)) {
               error = DELETEDMESSAGE;
-            tao
+          } else if (localId == null && EventLog.getInstance().isDeleted(pid.getValue())) {
               error = DELETEDMESSAGE;
             tao
           throw new NotFound("4874", "Couldn't find an object identified by " + pid.getValue()+". "+error);
             cjones
             cjones
-            cjones
+    } catch (RuntimeException e) {
-            cjones
+    	  ServiceFailure sf = new ServiceFailure("4872",
                 "Runtime Exception: Couldn't determine if node is allowed: " +
-            leinfelder
+                e.getMessage());
-            cjones
+    	  sf.initCause(e);
-            leinfelder
+        throw sf;
             cjones
             cjones
     return isAllowed;
             cjones
             cjones
-            cjones
+  /**
-            cjones
+   * Adds a new object to the Node, where the object is a science metadata object.
             cjones
    * @param session - the Session object containing the credentials for the Subject
    * @param pid - The object identifier to be created
    * @param object - the object bytes
    * @param sysmeta - the system metadata that describes the object
+   *
    * @return pid - the object identifier created
+   *
    * @throws InvalidToken
    * @throws ServiceFailure
    * @throws NotAuthorized
    * @throws IdentifierNotUnique
    * @throws UnsupportedType
    * @throws InsufficientResources
    * @throws InvalidSystemMetadata
    * @throws NotImplemented
    * @throws InvalidRequest
    */
   public Identifier create(Session session, Identifier pid, InputStream object,
     SystemMetadata sysmeta)
     throws InvalidToken, ServiceFailure, NotAuthorized, IdentifierNotUnique,
     UnsupportedType, InsufficientResources, InvalidSystemMetadata,
     NotImplemented, InvalidRequest {
             tao
    // verify the pid is valid format
       if (!isValidIdentifier(pid)) {
           throw new InvalidRequest("4891", "The provided identifier is invalid.");
+      }
-            cjones
+      // The lock to be used for this identifier
-            cjones
+      Lock lock = null;
             cjones
-            cjones
+      try {
-            cjones
+          lock = HazelcastService.getInstance().getLock(pid.getValue());
-            tao
+          lock.lock();
-            cjones
+          // are we allowed?
-            cjones
+          boolean isAllowed = false;
-            leinfelder
+          isAllowed = isAdminAuthorized(session);
             tao
           // additional check if it is the authoritative node if it is not the admin
           if(!isAllowed) {
               isAllowed = isAuthoritativeMNodeAdmin(session, pid);
+          }
             cjones
           // proceed if we're called by a CN
           if ( isAllowed ) {
-            tao
+              //check if the series id is legitimate. It uses the same rules of the method registerSystemMetadata
-            tao
+              //checkSidInModifyingSystemMetadata(sysmeta, "4896", "4893");
               Identifier sid = sysmeta.getSeriesId();
               if(sid != null) {
                   if (!isValidIdentifier(sid)) {
                       throw new InvalidRequest("4891", "The series id in the system metadata is invalid in the request.");
+                  }
+              }
-            cjones
+              // create the coordinating node version of the document
-            cjones
+              logMetacat.debug("Locked identifier " + pid.getValue());
-            cjones
+              sysmeta.setSerialVersion(BigInteger.ONE);
-            tao
+              //for the object whose authoritative mn is v1. we need reset the modification date.
               //for the object whose authoritative mn is v2. we just accept the modification date.
               D1NodeVersionChecker checker = new D1NodeVersionChecker(sysmeta.getAuthoritativeMemberNode());
               String version = checker.getVersion("MNStorage");
               if(version != null && version.equalsIgnoreCase(D1NodeVersionChecker.V1)) {
                   sysmeta.setDateSysMetadataModified(Calendar.getInstance().getTime());
+              }
-            leinfelder
+              //sysmeta.setArchived(false); // this is a create op, not update
             cjones
               // the CN should have set the origin and authoritative member node fields
               try {
                   sysmeta.getOriginMemberNode().getValue();
                   sysmeta.getAuthoritativeMemberNode().getValue();
               } catch (NullPointerException npe) {
                   throw new InvalidSystemMetadata("4896",
                       "Both the origin and authoritative member node identifiers need to be set.");
+              }
-            cjones
+              pid = super.create(session, pid, object, sysmeta);
           } else {
               String msg = "The subject listed as " + session.getSubject().getValue() +
                   " isn't allowed to call create() on a Coordinating Node.";
               logMetacat.info(msg);
               throw new NotAuthorized("1100", msg);
+          }
-            cjones
+      } catch (RuntimeException e) {
-            cjones
+          // Convert Hazelcast runtime exceptions to service failures
           String msg = "There was a problem creating the object identified by " +
               pid.getValue() + ". There error message was: " + e.getMessage();
-            cjones
+          throw new ServiceFailure("4893", msg);
             cjones
       } finally {
-            leinfelder
+    	  if (lock != null) {
 	          lock.unlock();
 	          logMetacat.debug("Unlocked identifier " + pid.getValue());
+    	  }
             cjones
-            cjones
+      return pid;
+  }
-            cjones
+  /**
    * Set access for a given object using the object identifier and a Subject
    * under a given Session.
-            tao
+   * This method only applies the objects whose authoritative mn is a v1 node.
-            cjones
+   * @param session - the Session object containing the credentials for the Subject
    * @param pid - the object identifier for the given object to apply the policy
    * @param policy - the access policy to be applied
+   *
    * @return true if the application of the policy succeeds
    * @throws InvalidToken
    * @throws ServiceFailure
    * @throws NotFound
    * @throws NotAuthorized
    * @throws NotImplemented
    * @throws InvalidRequest
    */
   public boolean setAccessPolicy(Session session, Identifier pid,
-            cjones
+      AccessPolicy accessPolicy, long serialVersion)
-            cjones
+      throws InvalidToken, ServiceFailure, NotFound, NotAuthorized,
-            cjones
+      NotImplemented, InvalidRequest, VersionMismatch {
             cjones
-            tao
+   // do we have a valid pid?
       if (pid == null || pid.getValue().trim().equals("")) {
           throw new InvalidRequest("4402", "The provided identifier was invalid.");
+      }
       String serviceFailureCode = "4430";
       Identifier sid = getPIDForSID(pid, serviceFailureCode);
       if(sid != null) {
           pid = sid;
+      }
-            cjones
+      // The lock to be used for this identifier
-            cjones
+      Lock lock = null;
-            cjones
+      SystemMetadata systemMetadata = null;
             cjones
-            cjones
+      boolean success = false;
       // get the subject
       Subject subject = session.getSubject();
-            cjones
+      // are we allowed to do this?
       if (!isAuthorized(session, pid, Permission.CHANGE_PERMISSION)) {
           throw new NotAuthorized("4420", "not allowed by "
                   + subject.getValue() + " on " + pid.getValue());
             cjones
       try {
-            cjones
+          lock = HazelcastService.getInstance().getLock(pid.getValue());
-            cjones
+          lock.lock();
-            cjones
+          logMetacat.debug("Locked identifier " + pid.getValue());
-            cjones
+          try {
               systemMetadata = HazelcastService.getInstance().getSystemMetadataMap().get(pid);
             cjones
-            cjones
+              if ( systemMetadata == null ) {
                   throw new NotFound("4400", "Couldn't find an object identified by " + pid.getValue());
+              }
-            cjones
+              // 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.";
-            cjones
+                 throw new VersionMismatch("4402", msg);
             cjones
-            tao
+              D1NodeVersionChecker checker = new D1NodeVersionChecker(systemMetadata.getAuthoritativeMemberNode());
               String version = checker.getVersion("MNStorage");
               if(version == null) {
-            tao
+                  throw new ServiceFailure("4430", "Couldn't determine the authoritative member node storage version for the pid "+pid.getValue());
-            tao
+              } else if (version.equalsIgnoreCase(D1NodeVersionChecker.V2)) {
                   //we don't apply this method to an object whose authoritative node is v2
                   throw new NotAuthorized("4420", V2V1MISSMATCH);
               } else if (!version.equalsIgnoreCase(D1NodeVersionChecker.V1)) {
                   //we don't understand this version (it is not v1 or v2)
                   throw new InvalidRequest("4402", "The version of the MNStorage is "+version+" for the authoritative member node of the object "+pid.getValue()+". We don't support it.");
+              }
-            cjones
+          } catch (RuntimeException e) {
-            cjones
+              // convert Hazelcast RuntimeException to NotFound
               throw new NotFound("4400", "No record found for: " + pid);
             cjones
             cjones
           // set the access policy
           systemMetadata.setAccessPolicy(accessPolicy);
             cjones
-            cjones
+          // update the system metadata
           try {
               systemMetadata.setSerialVersion(systemMetadata.getSerialVersion().add(BigInteger.ONE));
               systemMetadata.setDateSysMetadataModified(Calendar.getInstance().getTime());
               HazelcastService.getInstance().getSystemMetadataMap().put(systemMetadata.getIdentifier(), systemMetadata);
-            cjones
+              notifyReplicaNodes(systemMetadata);
-            cjones
+          } catch (RuntimeException e) {
-            cjones
+              // convert Hazelcast RuntimeException to ServiceFailure
               throw new ServiceFailure("4430", e.getMessage());
+          }
             cjones
-            cjones
+      } catch (RuntimeException e) {
-            cjones
+          throw new ServiceFailure("4430", e.getMessage());
             cjones
-            cjones
+      } finally {
-            cjones
+          lock.unlock();
-            cjones
+          logMetacat.debug("Unlocked identifier " + pid.getValue());
             cjones
+      }
             cjones
             cjones
     // TODO: how do we know if the map was persisted?
     success = true;
     return success;
+  }
-            cjones
+  /**
    * Full replacement of replication metadata in the system metadata for the
    * specified object, changes date system metadata modified
+   *
    * @param session - the Session object containing the credentials for the Subject
    * @param pid - the object identifier for the given object to apply the policy
    * @param replica - the replica to be updated
    * @return
    * @throws NotImplemented
    * @throws NotAuthorized
    * @throws ServiceFailure
    * @throws InvalidRequest
    * @throws NotFound
-            cjones
+   * @throws VersionMismatch
-            cjones
+   */
-            cjones
+  @Override
-            cjones
+  public boolean updateReplicationMetadata(Session session, Identifier pid,
-            cjones
+      Replica replica, long serialVersion)
-            cjones
+      throws NotImplemented, NotAuthorized, ServiceFailure, InvalidRequest,
-            cjones
+      NotFound, VersionMismatch {
             cjones
-            cjones
+      // The lock to be used for this identifier
-            cjones
+      Lock lock = null;
             cjones
-            cjones
+      // get the subject
       Subject subject = session.getSubject();
       // are we allowed to do this?
-            tao
+      if(session == null) {
           throw new NotAuthorized("4851", "Session cannot be null. It is not authorized for updating the replication metadata of the object "+pid.getValue());
       } else {
           if(!isCNAdmin(session)) {
               throw new NotAuthorized("4851", "The client -"+ session.getSubject().getValue()+ "is not a CN and is not authorized for updating the replication metadata of the object "+pid.getValue());
+        }
+      }
       /*try {
             cjones
           // what is the controlling permission?
           if (!isAuthorized(session, pid, Permission.WRITE)) {
               throw new NotAuthorized("4851", "not allowed by "
                       + subject.getValue() + " on " + pid.getValue());
+          }
             cjones
       } catch (InvalidToken e) {
           throw new NotAuthorized("4851", "not allowed by " + subject.getValue() +
                   " on " + pid.getValue());
-            tao
+      }*/
             cjones
       SystemMetadata systemMetadata = null;
-            cjones
+      try {
-            cjones
+          lock = HazelcastService.getInstance().getLock(pid.getValue());
-            cjones
+          lock.lock();
-            cjones
+          logMetacat.debug("Locked identifier " + pid.getValue());
             cjones
-            cjones
+          try {
               systemMetadata = HazelcastService.getInstance().getSystemMetadataMap().get(pid);
               // 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.";
-            cjones
+                 throw new VersionMismatch("4855", msg);
             cjones
-            cjones
+          } catch (RuntimeException e) { // Catch is generic since HZ throws RuntimeException
               throw new NotFound("4854", "No record found for: " + pid.getValue() +
                   " : " + e.getMessage());
             cjones
             cjones
             cjones
           // set the status for the replica
           List<Replica> replicas = systemMetadata.getReplicaList();
           NodeReference replicaNode = replica.getReplicaMemberNode();
-            cjones
+          ReplicationStatus replicaStatus = replica.getReplicationStatus();
-            cjones
+          int index = 0;
           for (Replica listedReplica: replicas) {
               // remove the replica that we are replacing
               if ( replicaNode.getValue().equals(listedReplica.getReplicaMemberNode().getValue())) {
-            cjones
+                      // don't allow status to change from COMPLETED to anything other
                       // than INVALIDATED: prevents overwrites from race conditions
-            cjones
+                	  if ( !listedReplica.getReplicationStatus().equals(replicaStatus) &&
                 	       listedReplica.getReplicationStatus().equals(ReplicationStatus.COMPLETED) &&
             		       !replicaStatus.equals(ReplicationStatus.INVALIDATED) ) {
-            cjones
+                	  throw new InvalidRequest("4853", "Status state change from " +
                 			  listedReplica.getReplicationStatus() + " to " +
                 			  replicaStatus.toString() + "is prohibited for identifier " +
                 			  pid.getValue() + " and target node " +
                 			  listedReplica.getReplicaMemberNode().getValue());
+            	  }
-            cjones
+                  replicas.remove(index);
                   break;
+              }
               index++;
+          }
             cjones
-            cjones
+          // add the new replica item
           replicas.add(replica);
           systemMetadata.setReplicaList(replicas);
             cjones
-            cjones
+          // update the metadata
           try {
               systemMetadata.setSerialVersion(systemMetadata.getSerialVersion().add(BigInteger.ONE));
-            cjones
+              // Based on CN behavior discussion 9/16/15, we no longer want to
               // update the modified date for changes to the replica list
               //systemMetadata.setDateSysMetadataModified(Calendar.getInstance().getTime());
-            cjones
+              HazelcastService.getInstance().getSystemMetadataMap().put(systemMetadata.getIdentifier(), systemMetadata);
             cjones
               // inform replica nodes of the change if the status is complete
-            cjones
+              if ( replicaStatus.equals(ReplicationStatus.COMPLETED) ) {
-            cjones
+            	  notifyReplicaNodes(systemMetadata);
             cjones
+              }
-            cjones
+          } catch (RuntimeException e) {
               logMetacat.info("Unknown RuntimeException thrown: " + e.getCause().getMessage());
-            cjones
+              throw new ServiceFailure("4852", e.getMessage());
             cjones
+          }
             cjones
-            cjones
+      } catch (RuntimeException e) {
           logMetacat.info("Unknown RuntimeException thrown: " + e.getCause().getMessage());
           throw new ServiceFailure("4852", e.getMessage());
       } finally {
           lock.unlock();
           logMetacat.debug("Unlocked identifier " + pid.getValue());
+      }
             cjones
       return true;
+  }
             leinfelder
-            cjones
+  /**
+   *
    */
   @Override
-            cjones
+  public ObjectList listObjects(Session session, Date startTime,
-            tao
+      Date endTime, ObjectFormatIdentifier formatid, NodeReference nodeId,Identifier identifier,
-            cjones
+      Integer start, Integer count)
-            cjones
+      throws InvalidRequest, InvalidToken, NotAuthorized, NotImplemented,
       ServiceFailure {
             tao
       return super.listObjects(session, startTime, endTime, formatid, identifier, nodeId, start, count);
             cjones
             leinfelder
             cjones
  	/**
  	 * Returns a list of checksum algorithms that are supported by DataONE.
  	 * @return cal  the list of checksum algorithms
+ 	 *
  	 * @throws ServiceFailure
  	 * @throws NotImplemented
  	 */
-            cjones
+  @Override
-            cjones
+  public ChecksumAlgorithmList listChecksumAlgorithms()
-            leinfelder
+			throws ServiceFailure, NotImplemented {
 		ChecksumAlgorithmList cal = new ChecksumAlgorithmList();
 		cal.addAlgorithm("MD5");
 		cal.addAlgorithm("SHA-1");
-            cjones
+		return cal;
             leinfelder
             cjones
   /**
    * Notify replica Member Nodes of system metadata changes for a given pid
+   *
    * @param currentSystemMetadata - the up to date system metadata
    */
   public void notifyReplicaNodes(SystemMetadata currentSystemMetadata) {
       Session session = null;
       List<Replica> replicaList = currentSystemMetadata.getReplicaList();
-            tao
+      //MNode mn = null;
-            cjones
+      NodeReference replicaNodeRef = null;
       CNode cn = null;
       NodeType nodeType = null;
       List<Node> nodeList = null;
       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 system metadata changes due " +
               "to communication issues with the CN: " + e.getMessage());
+      }
       if ( replicaList != null ) {
           // iterate through the replicas and inform  MN replica nodes
           for (Replica replica : replicaList) {
-            tao
+              String replicationVersion = null;
-            cjones
+              replicaNodeRef = replica.getReplicaMemberNode();
               try {
                   if (nodeList != null) {
                       // find the node type
                       for (Node node : nodeList) {
-            cjones
+                          if ( node.getIdentifier().getValue().equals(replicaNodeRef.getValue()) ) {
-            cjones
+                              nodeType = node.getType();
-            tao
+                              D1NodeVersionChecker checker = new D1NodeVersionChecker(replicaNodeRef);
                               replicationVersion = checker.getVersion("MNRead");
-            cjones
+                              break;
+                          }
+                      }
+                  }
                   // notify only MNs
-            tao
+                  if (replicationVersion != null && nodeType != null && nodeType == NodeType.MN) {
                       if(replicationVersion.equalsIgnoreCase(D1NodeVersionChecker.V2)) {
                           //connect to a v2 mn
                           MNode mn = D1Client.getMN(replicaNodeRef);
                           mn.systemMetadataChanged(session,
                               currentSystemMetadata.getIdentifier(),
                               currentSystemMetadata.getSerialVersion().longValue(),
                               currentSystemMetadata.getDateSysMetadataModified());
                       } else if (replicationVersion.equalsIgnoreCase(D1NodeVersionChecker.V1)) {
                           //connect to a v1 mn
                           org.dataone.client.v1.MNode mn = org.dataone.client.v1.itk.D1Client.getMN(replicaNodeRef);
                           mn.systemMetadataChanged(session,
                                   currentSystemMetadata.getIdentifier(),
                                   currentSystemMetadata.getSerialVersion().longValue(),
                                   currentSystemMetadata.getDateSysMetadataModified());
+                      }
             cjones
               } catch (Exception e) { // handle BaseException and other I/O issues
                   // swallow errors since the call is not critical
                   logMetacat.error("Can't inform "
                           + replicaNodeRef.getValue()
                           + " of system metadata changes due "
                           + "to communication issues with the CN: "
                           + e.getMessage());
+              }
+          }
+      }
+  }
             tao
-            tao
+  /**
    * Update the system metadata of the specified pid.
-            tao
+   * Note: the serial version and the replica list in the new system metadata will be ignored and the old values will be kept.
-            tao
+   */
   @Override
   public boolean updateSystemMetadata(Session session, Identifier pid,
           SystemMetadata sysmeta) throws NotImplemented, NotAuthorized,
           ServiceFailure, InvalidRequest, InvalidSystemMetadata, InvalidToken {
    if(sysmeta == null) {
        throw  new InvalidRequest("4863", "The system metadata object should NOT be null in the updateSystemMetadata request.");
+   }
    if(pid == null || pid.getValue() == null) {
        throw new InvalidRequest("4863", "Please specify the id in the updateSystemMetadata request ") ;
+   }
    if (session == null) {
        //TODO: many of the thrown exceptions do not use the correct error codes
        //check these against the docs and correct them
        throw new NotAuthorized("4861", "No Session - could not authorize for updating system metadata." +
                "  If you are not logged in, please do so and retry the request.");
    } else {
          //only CN is allwoed
          if(!isCNAdmin(session)) {
                throw new NotAuthorized("4861", "The client -"+ session.getSubject().getValue()+ "is not authorized for updating the system metadata of the object "+pid.getValue());
+         }
+   }
             tao
     //update the system metadata locally
     boolean success = false;
     try {
         HazelcastService.getInstance().getSystemMetadataMap().lock(pid);
         SystemMetadata currentSysmeta = HazelcastService.getInstance().getSystemMetadataMap().get(pid);
             tao
-            tao
+        if(currentSysmeta == null) {
             throw  new InvalidRequest("4863", "We can't find the current system metadata on the member node for the id "+pid.getValue());
+        }
-            tao
+        // CN will ignore the comming serial version and replica list fields from the mn node.
         BigInteger currentSerialVersion = currentSysmeta.getSerialVersion();
         sysmeta.setSerialVersion(currentSerialVersion);
         List<Replica> replicas = currentSysmeta.getReplicaList();
         sysmeta.setReplicaList(replicas);
-            tao
+        boolean needUpdateModificationDate = false;//cn doesn't need to change the modification date.
-            tao
+        boolean fromCN = true;
         success = updateSystemMetadata(session, pid, sysmeta, needUpdateModificationDate, currentSysmeta, fromCN);
-            tao
+    } finally {
         HazelcastService.getInstance().getSystemMetadataMap().unlock(pid);
+    }
-            tao
+    return success;
+  }
-            tao
+    @Override
     public boolean synchronize(Session session, Identifier pid) throws NotAuthorized, InvalidRequest, NotImplemented{
         throw new NotImplemented("0000", "CN query services are not implemented in Metacat.");
             tao
             leinfelder
 	@Override
-            leinfelder
+	public QueryEngineDescription getQueryEngineDescription(Session session,
 			String queryEngine) throws InvalidToken, ServiceFailure, NotAuthorized,
 			NotImplemented, NotFound {
 		throw new NotImplemented("0000", "CN query services are not implemented in Metacat.");
             leinfelder
+	}
 	@Override
-            leinfelder
+	public QueryEngineList listQueryEngines(Session session) throws InvalidToken,
 			ServiceFailure, NotAuthorized, NotImplemented {
 		throw new NotImplemented("0000", "CN query services are not implemented in Metacat.");
             leinfelder
+	}
 	@Override
-            leinfelder
+	public InputStream query(Session session, String queryEngine, String query)
 			throws InvalidToken, ServiceFailure, NotAuthorized, InvalidRequest,
 			NotImplemented, NotFound {
 		throw new NotImplemented("0000", "CN query services are not implemented in Metacat.");
             leinfelder
+	}
 	@Override
-            leinfelder
+	public Node getCapabilities() throws NotImplemented, ServiceFailure {
 		throw new NotImplemented("0000", "The CN capabilities are not stored in Metacat.");
             leinfelder
             tao
             cjones

Project

General

Profile

Metacat