Metacat

/**

 *  '$RCSfile$'

 *    Purpose: A Class that implements replication for metacat

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

 *             National Center for Ecological Analysis and Synthesis

 *    Authors: Chad Berkley

 *

 *   '$Author: leinfelder $'

 *     '$Date: 2014-02-10 11:52:33 -0800 (Mon, 10 Feb 2014) $'

 * '$Revision: 8585 $'

 *

 * 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.replication;

import java.io.ByteArrayInputStream;

import java.io.ByteArrayOutputStream;

import java.io.File;

import java.io.FileInputStream;

import java.io.FileNotFoundException;

import java.io.FileOutputStream;

import java.io.IOException;

import java.io.InputStream;

import java.io.InputStreamReader;

import java.io.OutputStream;

import java.io.StringReader;

import java.io.Writer;

import java.net.MalformedURLException;

import java.net.URL;

import java.security.PrivateKey;

import java.security.cert.X509Certificate;

import java.sql.PreparedStatement;

import java.sql.ResultSet;

import java.sql.SQLException;

import java.sql.Timestamp;

import java.util.ArrayList;

import java.util.Arrays;

import java.util.Calendar;

import java.util.Date;

import java.util.Hashtable;

import java.util.List;

import java.util.Timer;

import java.util.Vector;

import javax.servlet.http.HttpServletRequest;

import javax.servlet.http.HttpServletResponse;

import org.apache.commons.io.IOUtils;

import org.apache.http.HttpResponse;

import org.apache.http.conn.scheme.Scheme;

import org.apache.http.conn.ssl.SSLSocketFactory;

import org.apache.log4j.Logger;

import org.dataone.client.RestClient;

import org.dataone.client.auth.CertificateManager;

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

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

import org.dataone.service.util.DateTimeMarshaller;

import org.dataone.service.util.TypeMarshaller;

import org.jibx.runtime.JiBXException;

import org.xml.sax.InputSource;

import org.xml.sax.SAXException;

import org.xml.sax.XMLReader;

import edu.ucsb.nceas.metacat.DocumentImpl;

import edu.ucsb.nceas.metacat.DocumentImplWrapper;

import edu.ucsb.nceas.metacat.EventLog;

import edu.ucsb.nceas.metacat.IdentifierManager;

import edu.ucsb.nceas.metacat.McdbDocNotFoundException;

import edu.ucsb.nceas.metacat.McdbException;

import edu.ucsb.nceas.metacat.accesscontrol.AccessControlException;

import edu.ucsb.nceas.metacat.accesscontrol.AccessControlForSingleFile;

import edu.ucsb.nceas.metacat.accesscontrol.PermOrderException;

import edu.ucsb.nceas.metacat.admin.upgrade.RemoveInvalidReplicas;

import edu.ucsb.nceas.metacat.admin.upgrade.dataone.GenerateORE;

import edu.ucsb.nceas.metacat.admin.upgrade.dataone.GenerateSystemMetadata;

import edu.ucsb.nceas.metacat.client.InsufficientKarmaException;

import edu.ucsb.nceas.metacat.database.DBConnection;

import edu.ucsb.nceas.metacat.database.DBConnectionPool;

import edu.ucsb.nceas.metacat.database.DatabaseService;

import edu.ucsb.nceas.metacat.dataone.SyncAccessPolicy;

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

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

import edu.ucsb.nceas.metacat.properties.PropertyService;

import edu.ucsb.nceas.metacat.shared.BaseService;

import edu.ucsb.nceas.metacat.shared.HandlerException;

import edu.ucsb.nceas.metacat.shared.ServiceException;

import edu.ucsb.nceas.metacat.util.DocumentUtil;

import edu.ucsb.nceas.metacat.util.MetacatUtil;

import edu.ucsb.nceas.metacat.util.ReplicationUtil;

import edu.ucsb.nceas.metacat.util.SystemUtil;

import edu.ucsb.nceas.utilities.FileUtil;

import edu.ucsb.nceas.utilities.GeneralPropertyException;

import edu.ucsb.nceas.utilities.PropertyNotFoundException;

import edu.ucsb.nceas.utilities.access.DocInfoHandler;

import edu.ucsb.nceas.utilities.access.XMLAccessDAO;

public class ReplicationService extends BaseService {

	private static ReplicationService replicationService = null;

	private long timeInterval;

	private Date firstTime;

	private boolean timedReplicationIsOn = false;

	Timer replicationDaemon;

	private static Vector<String> fileLocks = new Vector<String>();

//	private Thread lockThread = null;

	public static final String FORCEREPLICATEDELETE = "forcereplicatedelete";

	public static final String FORCEREPLICATEDELETEALL = "forcereplicatedeleteall";

	private static String TIMEREPLICATION = "replication.timedreplication";

	private static String TIMEREPLICATIONINTERVAl ="replication.timedreplicationinterval";

	private static String FIRSTTIME = "replication.firsttimedreplication";

	private static final int TIMEINTERVALLIMIT = 7200000;

	public static final String REPLICATIONUSER = "replication";

	private static int CLIENTTIMEOUT = 30000;

	public static final String REPLICATION_LOG_FILE_NAME = "metacatreplication.log";

	private static String DATA_FILE_FLAG = null;

	public static String METACAT_REPL_ERROR_MSG = null;

	private static Logger logReplication = Logger.getLogger("ReplicationLogging");

	private static Logger logMetacat = Logger.getLogger(ReplicationService.class);

	static {

		// lookup the client timeout

		String clientTimeout = null;

		try {

			clientTimeout = PropertyService.getProperty("replication.client.timeout");

			CLIENTTIMEOUT = Integer.parseInt(clientTimeout);

		} catch (Exception e) {

			// just use default

			logReplication.warn("No custom client timeout specified in configuration, using default." + e.getMessage());

		}

		try {

			DATA_FILE_FLAG = PropertyService.getProperty("replication.datafileflag");

		} catch (PropertyNotFoundException e) {

			logReplication.error("No 'replication.datafileflag' specified in configuration." + e.getMessage());

		}

	}

	private ReplicationService() throws ServiceException {

		_serviceName = "ReplicationService";

		initialize();

	}

	private void initialize() throws ServiceException {

		// initialize db connections to handle any update requests

		// deltaT = util.getProperty("replication.deltaT");

		// the default deltaT can be set from metacat.properties

		// create a thread to do the delta-T check but don't execute it yet

		replicationDaemon = new Timer(true);

		try {

			String replLogFile = PropertyService.getProperty("replication.logdir")

				+ FileUtil.getFS() + REPLICATION_LOG_FILE_NAME;

			METACAT_REPL_ERROR_MSG = "An error occurred in replication.  Please see the " +

				"replication log at: " + replLogFile;

			String timedRepIsOnStr = 

				PropertyService.getProperty("replication.timedreplication");

			timedReplicationIsOn = (new Boolean(timedRepIsOnStr)).booleanValue();

			logReplication.info("ReplicationService.initialize - The timed replication on is" + timedReplicationIsOn);

			String timeIntervalStr = 

				PropertyService.getProperty("replication.timedreplicationinterval");

			timeInterval = (new Long(timeIntervalStr)).longValue();

			logReplication.info("ReplicationService.initialize - The timed replication time Interval is " + timeInterval);

			String firstTimeStr = 

				PropertyService.getProperty("replication.firsttimedreplication");

			logReplication.info("ReplicationService.initialize - first replication time form property is " + firstTimeStr);

			firstTime = ReplicationHandler.combinateCurrentDateAndGivenTime(firstTimeStr);

			logReplication.info("ReplicationService.initialize - After combine current time, the real first time is "

					+ firstTime.toString() + " minisec");

			// set up time replication if it is on

			if (timedReplicationIsOn) {

				replicationDaemon.scheduleAtFixedRate(new ReplicationHandler(),

						firstTime, timeInterval);

				logReplication.info("ReplicationService.initialize - deltaT handler started with rate="

						+ timeInterval + " mini seconds at " + firstTime.toString());

			}

		} catch (PropertyNotFoundException pnfe) {

			throw new ServiceException(

					"ReplicationService.initialize - Property error while instantiating "

							+ "replication service: " + pnfe.getMessage());

		} catch (HandlerException he) {

			throw new ServiceException(

					"ReplicationService.initialize - Handler error while instantiating "

							+ "replication service" + he.getMessage());

		} 

	}

	/**

	 * Get the single instance of SessionService.

	 * 

	 * @return the single instance of SessionService

	 */

	public static ReplicationService getInstance() throws ServiceException {

		if (replicationService == null) {

			replicationService = new ReplicationService();

		}

		return replicationService;

	}

	public boolean refreshable() {

		return true;

	}

	protected void doRefresh() throws ServiceException {

		return;

	}

	public void stop() throws ServiceException{

	}

	public void stopReplication() throws ServiceException {

	      //stop the replication server

	      replicationDaemon.cancel();

	      replicationDaemon = new Timer(true);

	      timedReplicationIsOn = false;

	      try {

	    	  PropertyService.setProperty("replication.timedreplication", (new Boolean(timedReplicationIsOn)).toString());

	      } catch (GeneralPropertyException gpe) {

	    	  logReplication.warn("ReplicationService.stopReplication - Could not set replication.timedreplication property: " + gpe.getMessage());

	      }

	      logReplication.info("ReplicationService.stopReplication - deltaT handler stopped");

		return;

	}

	public void startReplication(Hashtable<String, String[]> params) throws ServiceException {

	       String firstTimeStr = "";

	      //start the replication server

	       if ( params.containsKey("rate") ) {

	        timeInterval = new Long(

	               new String(((String[])params.get("rate"))[0])).longValue();

	        if(timeInterval < TIMEINTERVALLIMIT) {

	            //deltaT<30 is a timing mess!

	            timeInterval = TIMEINTERVALLIMIT;

	            throw new ServiceException("Replication deltaT rate cannot be less than "+

	                    TIMEINTERVALLIMIT + " millisecs and system automatically setup the rate to "+TIMEINTERVALLIMIT);

	        }

	      } else {

	        timeInterval = TIMEINTERVALLIMIT ;

	      }

	      logReplication.info("ReplicationService.startReplication - New rate is: " + timeInterval + " mini seconds.");

	      if ( params.containsKey("firsttime"))

	      {

	         firstTimeStr = ((String[])params.get("firsttime"))[0];

	         try

	         {

	           firstTime = ReplicationHandler.combinateCurrentDateAndGivenTime(firstTimeStr);

	           logReplication.info("ReplicationService.startReplication - The first time setting is "+firstTime.toString());

	         }

	         catch (HandlerException e)

	         {

	            throw new ServiceException(e.getMessage());

	         }

	         logReplication.warn("After combine current time, the real first time is "

	                                  +firstTime.toString()+" minisec");

	      }

	      else

	      {

	    	  logMetacat.error("ReplicationService.startReplication - " + ReplicationService.METACAT_REPL_ERROR_MSG);                         

	          logReplication.error("ReplicationService.startReplication - You should specify the first time " +

	                                  "to start a time replication");

	          return;

	      }

	      timedReplicationIsOn = true;

	      try {

	      // save settings to property file

	      PropertyService.setProperty(TIMEREPLICATION, (new Boolean(timedReplicationIsOn)).toString());

	      // note we couldn't use firstTime object because it has date info

	      // we only need time info such as 10:00 PM

	      PropertyService.setProperty(FIRSTTIME, firstTimeStr);

	      PropertyService.setProperty(TIMEREPLICATIONINTERVAl, (new Long(timeInterval)).toString());

	      } catch (GeneralPropertyException gpe) {

	    	  logReplication.warn("ReplicationService.startReplication - Could not set property: " + gpe.getMessage());

	      }

	      replicationDaemon.cancel();

	      replicationDaemon = new Timer(true);

	      replicationDaemon.scheduleAtFixedRate(new ReplicationHandler(), firstTime,

	                                            timeInterval);

	      logReplication.info("ReplicationService.startReplication - deltaT handler started with rate=" +

	                                    timeInterval + " milliseconds at " +firstTime.toString());

	}

	public void runOnce() throws ServiceException {

	      //updates this server exactly once

	      replicationDaemon.schedule(new ReplicationHandler(), 0);

	}

	/**

	 * This method can add, delete and list the servers currently included in

	 * xml_replication.

	 * action           subaction            other needed params

	 * ---------------------------------------------------------

	 * servercontrol    add                  server

	 * servercontrol    delete               server

	 * servercontrol    list

	 */

	public static void handleServerControlRequest(

			Hashtable<String, String[]> params, HttpServletRequest request, HttpServletResponse response) {

		String subaction = ((String[]) params.get("subaction"))[0];

		DBConnection dbConn = null;

		int serialNumber = -1;

		PreparedStatement pstmt = null;

		String replicate = null;

		String server = null;

		String dataReplicate = null;

		String hub = null;

		Writer out = null;

		boolean showGenerateSystemMetadata = false;

		try {

			response.setContentType("text/xml");

			out = response.getWriter();

			//conn = util.openDBConnection();

			dbConn = DBConnectionPool

					.getDBConnection("MetacatReplication.handleServerControlRequest");

			serialNumber = dbConn.getCheckOutSerialNumber();

			// add server to server list

			if (subaction.equals("add")) {

				replicate = ((String[]) params.get("replicate"))[0];

				server = ((String[]) params.get("server"))[0];

				//Get data replication value

				dataReplicate = ((String[]) params.get("datareplicate"))[0];

				//Get hub value

				hub = ((String[]) params.get("hub"))[0];

				Calendar cal = Calendar.getInstance();

                cal.set(1980, 1, 1);

				String sql = "INSERT INTO xml_replication "

						+ "(server, last_checked, replicate, datareplicate, hub) "

						+ "VALUES (?,?,?,?,?)";

				pstmt = dbConn.prepareStatement(sql);

				pstmt.setString(1, server);

				pstmt.setTimestamp(2, new Timestamp(cal.getTimeInMillis()));

				pstmt.setInt(3, Integer.parseInt(replicate));

				pstmt.setInt(4, Integer.parseInt(dataReplicate));

				pstmt.setInt(5, Integer.parseInt(hub));

				String sqlReport = "XMLAccessAccess.getXMLAccessForDoc - SQL: " + sql;

				sqlReport += " [" + server + "," + replicate + 

					"," + dataReplicate + "," + hub + "]";

				logMetacat.info(sqlReport);

				pstmt.execute();

				pstmt.close();

				dbConn.commit();

				out.write("Server " + server + " added");

				// delete server from server list

			} else if (subaction.equals("delete")) {

				server = ((String[]) params.get("server"))[0];

				pstmt = dbConn.prepareStatement("DELETE FROM xml_replication "

						+ "WHERE server LIKE ?");

				pstmt.setString(1, server);

				pstmt.execute();

				pstmt.close();

				dbConn.commit();

				out.write("Server " + server + " deleted");

			} else if (subaction.equals("list")) {

				// nothing special - it's the default behavior

			} else if (subaction.equals("generatesystemmetadata")) {

				GenerateSystemMetadata gsm = new GenerateSystemMetadata();

				int serverLocation = -1;

				String serverid = ((String[]) params.get("serverid"))[0];

				serverLocation = Integer.parseInt(serverid);

				gsm.setServerLocation(serverLocation );

				gsm.multiThreadUpgrade();

				out.write("System Metadata generated for server " + serverid);

			} else if (subaction.equals("generateore")) {

				GenerateORE gore = new GenerateORE();

				int serverLocation = -1;

				String serverid = ((String[]) params.get("serverid"))[0];

				serverLocation = Integer.parseInt(serverid);

				gore.setServerLocation(serverLocation );

				gore.upgrade();

				out.write("Generated ORE maps for server " + serverid);

			} else if (subaction.equals("removeinvalidreplicas")) {

				RemoveInvalidReplicas rir = new RemoveInvalidReplicas();

				int serverLocation = -1;

				String serverid = ((String[]) params.get("serverid"))[0];

				serverLocation = Integer.parseInt(serverid);

				rir.setServerLocation(serverLocation );

				rir.upgrade();

				out.write("Removed invalid replicas for server " + serverid);

			} else if (subaction.equals("syncaccesspolicy")) {

				SyncAccessPolicy syncAP = new SyncAccessPolicy();

				response.setContentType("text/html");

				out = response.getWriter();

				if (params.containsKey("pid")) {

					String[] pids = params.get("pid");

					logMetacat.debug("Attempting to sync access policies for pids: " + pids);

					ArrayList<String> pidsToSync = new ArrayList<String>(Arrays.asList(pids));

					try {

						List<Identifier> syncedPids = syncAP.sync(pidsToSync);

						out.write("<html><body>Syncing access policies has completed for " + syncedPids.size() + " pids.</body></html>");

					} catch (Exception e) {

						logMetacat.error("Error syncing all access polies: "

								+ e.getMessage());

						response.setContentType("text/html");

						out = response.getWriter();

						out.write("<html><body>Error syncing access policies</body></html>");

					}

				} else {

					logMetacat.debug("Syncing access policies for all docids for this node");

					try {

						syncAP.syncAll();

						out.write("<html><body>Syncing access policies for all docids has completed.</body></html>");

					} catch (Exception e) {

						logMetacat.error("Error syncing access policies: "

								+ e.getMessage());

						out.write("<html><body>Error syncing access policies: " + e.getMessage() + " </body></html>");

					}

				}

			} else {

				out.write("<error>Unkonwn subaction</error>");

				return;

			}

			// show SM generate button?

			String dataoneConfigured = PropertyService.getProperty("configutil.dataoneConfigured");

			if (dataoneConfigured != null) {

				showGenerateSystemMetadata = Boolean.parseBoolean(dataoneConfigured);

			}

			// always list them after processing

			response.setContentType("text/html");

			out.write("<html><body><table border=\"1\">");

			out.write("<tr><td><b>server</b></td>");

			out.write("<td><b>last_checked</b></td>");

			out.write("<td><b>replicate</b></td>");

			out.write("<td><b>datareplicate</b></td>");

			out.write("<td><b>hub</b></td>");

			if (showGenerateSystemMetadata) {

				out.write("<td><b>System Metadata</b></td>");

				out.write("<td><b>ORE Maps</b></td>");

				out.write("<td><b>Invalid Replicas</b></td>");

			}

			out.write("<td><b>Sync Access Policies</b></td>");

			out.write("</tr>");

			pstmt = dbConn.prepareStatement("SELECT serverid, server, last_checked, replicate, datareplicate, hub FROM xml_replication");

			pstmt.execute();

			ResultSet rs = pstmt.getResultSet();

			boolean tablehasrows = rs.next();

			while (tablehasrows) {

				String serverId = rs.getString(1);

				out.write("<tr><td>" + rs.getString(2) + "</td><td>");

				out.write(rs.getString(3) + "</td><td>");

				out.write(rs.getString(4) + "</td><td>");

				out.write(rs.getString(5) + "</td><td>");

				out.write(rs.getString(6) + "</td>");

				if (showGenerateSystemMetadata) {

					// for SM

					out.write("<td><form action='" + request.getContextPath() + "/admin'>");

					out.write("<input name='serverid' type='hidden' value='" + serverId  + "'/>");

					out.write("<input name='configureType' type='hidden' value='replication'/>");

					out.write("<input name='action' type='hidden' value='servercontrol'/>");

					out.write("<input name='subaction' type='hidden' value='generatesystemmetadata'/>");

					out.write("<input type='submit' value='Generate System Metadata'/>");

					out.write("</form></td>");

					// for ORE maps

					out.write("<td><form action='" + request.getContextPath() + "/admin'>");

					out.write("<input name='serverid' type='hidden' value='" + serverId + "'/>");

					out.write("<input name='configureType' type='hidden' value='replication'/>");

					out.write("<input name='action' type='hidden' value='servercontrol'/>");

					out.write("<input name='subaction' type='hidden' value='generateore'/>");

					out.write("<input type='submit' value='Generate ORE'/>");

					out.write("</form></td>");

					// for invalid replicas

					out.write("<td><form action='" + request.getContextPath() + "/admin'>");

					out.write("<input name='serverid' type='hidden' value='" + serverId + "'/>");

					out.write("<input name='configureType' type='hidden' value='replication'/>");

					out.write("<input name='action' type='hidden' value='servercontrol'/>");

					out.write("<input name='subaction' type='hidden' value='removeinvalidreplicas'/>");

					String disabled = "";

					if (Integer.valueOf(serverId) == 1) {

						disabled = "disabled='true'";

					}

					out.write("<input type='submit' value='Remove Invalid Replicas' " + disabled + " />");

					out.write("</form></td>");

				}

				// for syncing access policies (MN -> CN)

				out.write("<td><form action='" + request.getContextPath() + "/admin'>");

				out.write("<input name='serverid' type='hidden' value='" + serverId + "'/>");

				out.write("<input name='configureType' type='hidden' value='replication'/>");

				out.write("<input name='action' type='hidden' value='servercontrol'/>");

				out.write("<input name='subaction' type='hidden' value='syncaccesspolicy'/>");

				out.write("<input type='submit' value='Sync access policies'/>");

				out.write("</form></td>");

				out.write("</tr>");

				tablehasrows = rs.next();

			}

			out.write("</table></body></html>");

			pstmt.close();

			//conn.close();

		} catch (Exception e) {

			logMetacat.error("ReplicationService.handleServerControlRequest - " + ReplicationService.METACAT_REPL_ERROR_MSG);                         

			logReplication.error("ReplicationService.handleServerControlRequest - Error in "

					+ "MetacatReplication.handleServerControlRequest " + e.getMessage());

			e.printStackTrace(System.out);

		} finally {

			try {

				pstmt.close();

			}//try

			catch (SQLException ee) {

				logMetacat.error("ReplicationService.handleServerControlRequest - " + ReplicationService.METACAT_REPL_ERROR_MSG);                         

				logReplication.error("ReplicationService.handleServerControlRequest - Error in MetacatReplication.handleServerControlRequest to close pstmt "

						+ ee.getMessage());

			}//catch

			finally {

				DBConnectionPool.returnDBConnection(dbConn, serialNumber);

			}//finally

			if (out != null) {

				try {

					out.close();

				} catch (IOException e) {

					logMetacat.error(e.getMessage(), e);

				}

			}

		}//finally

	}

	/**

	 * when a forcereplication request comes in, local host sends a read request

	 * to the requesting server (remote server) for the specified docid. Then

	 * store it in local database.

	 */

	protected static void handleForceReplicateRequest(

			Hashtable<String, String[]> params, HttpServletResponse response,

			HttpServletRequest request) {

		String server = ((String[]) params.get("server"))[0]; // the server that

		String docid = ((String[]) params.get("docid"))[0]; // sent the document

		String dbaction = "UPDATE"; // the default action is UPDATE

		//    boolean override = false;

		//    int serverCode = 1;

		DBConnection dbConn = null;

		int serialNumber = -1;

		String docName = null;

		try {

			//if the url contains a dbaction then the default action is overridden

			if (params.containsKey("dbaction")) {

				dbaction = ((String[]) params.get("dbaction"))[0];

				//serverCode = MetacatReplication.getServerCode(server);

				//override = true; //we are now overriding the default action

			}

			logReplication.info("ReplicationService.handleForceReplicateRequest - Force replication request from: " + server);

			logReplication.info("ReplicationService.handleForceReplicateRequest - Force replication docid: " + docid);

			logReplication.info("ReplicationService.handleForceReplicateRequest - Force replication action: " + dbaction);

			// sending back read request to remote server

			URL u = new URL("https://" + server + "?server="

					+ MetacatUtil.getLocalReplicationServerName() + "&action=read&docid="

					+ docid);

			String xmldoc = ReplicationService.getURLContent(u);

			// get the document info from server

			URL docinfourl = new URL("https://" + server + "?server="

					+ MetacatUtil.getLocalReplicationServerName()

					+ "&action=getdocumentinfo&docid=" + docid);

			String docInfoStr = ReplicationService.getURLContent(docinfourl);

			// strip out the system metadata portion

			String systemMetadataXML = ReplicationUtil.getSystemMetadataContent(docInfoStr);

			docInfoStr = ReplicationUtil.getContentWithoutSystemMetadata(docInfoStr);

			//dih is the parser for the docinfo xml format

			DocInfoHandler dih = new DocInfoHandler();

			XMLReader docinfoParser = ReplicationHandler.initParser(dih);

			docinfoParser.parse(new InputSource(new StringReader(docInfoStr)));

			//      Hashtable<String,Vector<AccessControlForSingleFile>> docinfoHash = dih.getDocInfo();

			Hashtable<String, String> docinfoHash = dih.getDocInfo();

			// Get home server of this docid

			String homeServer = (String) docinfoHash.get("home_server");

			// process system metadata

			if (systemMetadataXML != null) {

				SystemMetadata sysMeta = 

					TypeMarshaller.unmarshalTypeFromStream(

							SystemMetadata.class,

							new ByteArrayInputStream(systemMetadataXML.getBytes("UTF-8")));

				// need the guid-to-docid mapping

				boolean mappingExists = true;

		      	mappingExists = IdentifierManager.getInstance().mappingExists(sysMeta.getIdentifier().getValue());

		      	if (!mappingExists) {

		      		IdentifierManager.getInstance().createMapping(sysMeta.getIdentifier().getValue(), docid);

		      	}

				// save the system metadata

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

				// submit for indexing

                MetacatSolrIndex.getInstance().submit(sysMeta.getIdentifier(), sysMeta, null);

			}

			// dates

			String createdDateString = docinfoHash.get("date_created");

			String updatedDateString = docinfoHash.get("date_updated");

			Date createdDate = DateTimeMarshaller.deserializeDateToUTC(createdDateString);

			Date updatedDate = DateTimeMarshaller.deserializeDateToUTC(updatedDateString);

			logReplication.info("ReplicationService.handleForceReplicateRequest - homeServer: " + homeServer);

			// Get Document type

			String docType = (String) docinfoHash.get("doctype");

			logReplication.info("ReplicationService.handleForceReplicateRequest - docType: " + docType);

			String parserBase = null;

			// this for eml2 and we need user eml2 parser

			if (docType != null

					&& (docType.trim()).equals(DocumentImpl.EML2_0_0NAMESPACE)) {

				logReplication.warn("ReplicationService.handleForceReplicateRequest - This is an eml200 document!");

				parserBase = DocumentImpl.EML200;

			} else if (docType != null

					&& (docType.trim()).equals(DocumentImpl.EML2_0_1NAMESPACE)) {

				logReplication.warn("ReplicationService.handleForceReplicateRequest - This is an eml2.0.1 document!");

				parserBase = DocumentImpl.EML200;

			} else if (docType != null

					&& (docType.trim()).equals(DocumentImpl.EML2_1_0NAMESPACE)) {

				logReplication.warn("ReplicationService.handleForceReplicateRequest - This is an eml2.1.0 document!");

				parserBase = DocumentImpl.EML210;

			} else if (docType != null

					&& (docType.trim()).equals(DocumentImpl.EML2_1_1NAMESPACE)) {

				logReplication.warn("ReplicationService.handleForceReplicateRequest - This is an eml2.1.1 document!");

				parserBase = DocumentImpl.EML210;

			}

			logReplication.warn("ReplicationService.handleForceReplicateRequest - The parserBase is: " + parserBase);

			// Get DBConnection from pool

			dbConn = DBConnectionPool

					.getDBConnection("MetacatReplication.handleForceReplicateRequest");

			serialNumber = dbConn.getCheckOutSerialNumber();

			// write the document to local database

			DocumentImplWrapper wrapper = new DocumentImplWrapper(parserBase, false, false);

			//try this independently so we can set access even if the update action is invalid (i.e docid has not changed)

			try {

				wrapper.writeReplication(dbConn, xmldoc, null, null,

						dbaction, docid, null, null, homeServer, server, createdDate,

						updatedDate);

			} finally {

				//process extra access rules before dealing with the write exception (doc exist already)

				try {

		        	// check if we had a guid -> docid mapping

		        	String docidNoRev = DocumentUtil.getDocIdFromAccessionNumber(docid);

		        	int rev = DocumentUtil.getRevisionFromAccessionNumber(docid);

		        	IdentifierManager.getInstance().getGUID(docidNoRev, rev);

		        	// no need to create the mapping if we have it

		        } catch (McdbDocNotFoundException mcdbe) {

		        	// create mapping if we don't

		        	IdentifierManager.getInstance().createMapping(docid, docid);

		        }

		        Vector<XMLAccessDAO> accessControlList = dih.getAccessControlList();

		        if (accessControlList != null) {

		        	AccessControlForSingleFile acfsf = new AccessControlForSingleFile(docid);

		        	for (XMLAccessDAO xmlAccessDAO : accessControlList) {

		        		try {

			        		if (!acfsf.accessControlExists(xmlAccessDAO)) {

			        			acfsf.insertPermissions(xmlAccessDAO);

								logReplication.info("ReplicationService.handleForceReplicateRequest - document " + docid

										+ " permissions added to DB");

			        		}

		        		} catch (PermOrderException poe) {

		        			// this is problematic, but should not prevent us from replicating

		        			// see https://redmine.dataone.org/issues/2583

		        			String msg = "Could not insert access control for: " + docid + " Message: " + poe.getMessage();

		        			logMetacat.error(msg, poe);

		        			logReplication.error(msg, poe);

		        		}

		            }

		        }

		        // process the real owner and updater

				String user = (String) docinfoHash.get("user_owner");

				String updated = (String) docinfoHash.get("user_updated");

		        updateUserOwner(dbConn, docid, user, updated);

				logReplication.info("ReplicationService.handleForceReplicateRequest - document " + docid + " added to DB with "

						+ "action " + dbaction);

				EventLog.getInstance().log(request.getRemoteAddr(), request.getHeader("User-Agent"), REPLICATIONUSER, docid, dbaction);

			}

		} catch (SQLException sqle) {

			logMetacat.error("ReplicationService.handleForceReplicateRequest - " + ReplicationService.METACAT_REPL_ERROR_MSG);                         

			logReplication.error("ReplicationService.handleForceReplicateRequest - SQL error when adding doc " + docid + 

					" to DB with action " + dbaction + ": " + sqle.getMessage());

		} catch (MalformedURLException mue) {

			logMetacat.error("ReplicationService.handleForceReplicateRequest - " + ReplicationService.METACAT_REPL_ERROR_MSG);                         

			logReplication.error("ReplicationService.handleForceReplicateRequest - URL error when adding doc " + docid + 

					" to DB with action " + dbaction + ": " + mue.getMessage());

		} catch (SAXException se) {

			logMetacat.error("ReplicationService.handleForceReplicateRequest - " + ReplicationService.METACAT_REPL_ERROR_MSG);                         

			logReplication.error("ReplicationService.handleForceReplicateRequest - SAX parsing error when adding doc " + docid + 

					" to DB with action " + dbaction + ": " + se.getMessage());

		} catch (HandlerException he) {

			logMetacat.error("ReplicationService.handleForceReplicateRequest - " + ReplicationService.METACAT_REPL_ERROR_MSG);                         

			logReplication.error("ReplicationService.handleForceReplicateRequest - Handler error when adding doc " + docid + 

					" to DB with action " + dbaction + ": " + he.getMessage());

		} catch (IOException ioe) {

			logMetacat.error("ReplicationService.handleForceReplicateRequest - " + ReplicationService.METACAT_REPL_ERROR_MSG);                         

			logReplication.error("ReplicationService.handleForceReplicateRequest - I/O error when adding doc " + docid + 

					" to DB with action " + dbaction + ": " + ioe.getMessage());

		} catch (PermOrderException poe) {

			logMetacat.error("ReplicationService.handleForceReplicateRequest - " + ReplicationService.METACAT_REPL_ERROR_MSG);                         

			logReplication.error("ReplicationService.handleForceReplicateRequest - Permissions order error when adding doc " + docid + 

					" to DB with action " + dbaction + ": " + poe.getMessage());

		} catch (AccessControlException ace) {

			logMetacat.error("ReplicationService.handleForceReplicateRequest - " + ReplicationService.METACAT_REPL_ERROR_MSG);                         

			logReplication.error("ReplicationService.handleForceReplicateRequest - Permissions order error when adding doc " + docid + 

					" to DB with action " + dbaction + ": " + ace.getMessage());

		} catch (Exception e) {

			logMetacat.error("ReplicationService.handleForceReplicateRequest - " + ReplicationService.METACAT_REPL_ERROR_MSG);                         

			logReplication.error("ReplicationService.handleForceReplicateRequest - General error when adding doc " + docid + 

					" to DB with action " + dbaction + ": " + e.getMessage());

		} finally {

			// Return the checked out DBConnection

			DBConnectionPool.returnDBConnection(dbConn, serialNumber);

		}//finally

	}

	/*

	 * when a forcereplication delete request comes in, local host will delete this

	 * document

	 */

	protected static void handleForceReplicateDeleteRequest(

			Hashtable<String, String[]> params, HttpServletResponse response,

			HttpServletRequest request, boolean removeAll) {

		String server = ((String[]) params.get("server"))[0]; // the server that

		String docid = ((String[]) params.get("docid"))[0]; // sent the document

		try {

			logReplication.info("ReplicationService.handleForceReplicateDeleteRequest - force replication delete request from " + server);

			logReplication.info("ReplicationService.handleForceReplicateDeleteRequest - force replication delete docid " + docid);

			logReplication.info("ReplicationService.handleForceReplicateDeleteRequest - Force replication delete request from: " + server);

			logReplication.info("ReplicationService.handleForceReplicateDeleteRequest - Force replication delete docid: " + docid);

			DocumentImpl.delete(docid, null, null, server, removeAll);

			logReplication.info("ReplicationService.handleForceReplicateDeleteRequest - document " + docid + " was successfully deleted ");

			EventLog.getInstance().log(request.getRemoteAddr(), request.getHeader("User-Agent"), REPLICATIONUSER, docid,

					"delete");

			logReplication.info("ReplicationService.handleForceReplicateDeleteRequest - document " + docid + " was successfully deleted ");

		} catch (McdbDocNotFoundException e) {

			logMetacat.error("ReplicationService.handleForceReplicateDeleteRequest - " + ReplicationService.METACAT_REPL_ERROR_MSG);                         

			logReplication.error("document " + docid

					+ " failed to delete because " + e.getMessage());

			logReplication.error("ReplicationService.handleForceReplicateDeleteRequest - error: " + e.getMessage());

		} catch (InsufficientKarmaException e) {

			logMetacat.error("ReplicationService.handleForceReplicateDeleteRequest - " + ReplicationService.METACAT_REPL_ERROR_MSG);                         

			logReplication.error("document " + docid

					+ " failed to delete because " + e.getMessage());

			logReplication.error("ReplicationService.handleForceReplicateDeleteRequest - error: " + e.getMessage());