package eu.dnetlib.data.hadoop.blackboard;

import java.io.IOException;
import java.util.List;
import java.util.Map;
import java.util.Map.Entry;

import com.google.common.base.Joiner;
import com.google.common.collect.Iterables;
import com.google.common.collect.Lists;
import eu.dnetlib.data.hadoop.HadoopJob;
import eu.dnetlib.data.hadoop.mapreduce.MapreduceJobMonitor;
import eu.dnetlib.rmi.data.hadoop.ClusterName;
import eu.dnetlib.rmi.data.hadoop.HadoopJobType.AdminJobType;
import eu.dnetlib.rmi.data.hadoop.HadoopServiceException;
import eu.dnetlib.data.hadoop.utils.JobProfile;
import eu.dnetlib.data.hadoop.utils.ScanProperties;
import eu.dnetlib.enabling.tools.blackboard.BlackboardJob;
import org.apache.commons.lang3.StringUtils;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.mapreduce.CopyTable;
import org.apache.hadoop.mapred.JobConf;
import org.apache.hadoop.mapred.RunningJob;
import org.apache.hadoop.mapreduce.Job;

/**
 * The Class SubmitAdminJobAction.
 */
public class SubmitAdminJobAction extends SubmitMapreduceJobAction {

	/**
	 * logger.
	 */
	private static final Log log = LogFactory.getLog(SubmitAdminJobAction.class); // NOPMD by marko on 11/24/08 5:02 PM

	/**
	 * (non-Javadoc)
	 *
	 * @see
	 * eu.dnetlib.data.hadoop..blackboard.SubmitMapreduceJobAction#executeAsync(eu.dnetlib.enabling.tools.blackboard.BlackboardServerHandler,
	 * eu.dnetlib.enabling.tools.blackboard.BlackboardJob)
	 */
	@Override
	public void submit(final JobCompletion callback, final BlackboardJob bbJob, String jobName, JobProfile jobProfile) throws HadoopServiceException {
		final ClusterName clusterName = ClusterName.valueOf(bbJob.getParameters().get("cluster"));

		try {
			JobConf jobConf = prepareJob(getConf(clusterName), jobName, jobProfile, bbJob.getParameters());

			jobConf = initAdminJob(bbJob, jobName, jobProfile, jobConf);

			if (!hadoopClientMap.isMapreduceAvailable(clusterName))
				throw new HadoopServiceException("mapreduce not available for cluster: " + clusterName.toString());

			logJobDetails(jobConf);

			final RunningJob runningJob = hadoopClientMap.getJtClient(clusterName).submitJob(jobConf);
			final String jobId = newJobId(clusterName, runningJob.getID().getId());

			jobRegistry.registerJob(HadoopJob.newInstance(jobId, clusterName, jobProfile,
					new MapreduceJobMonitor(runningJob, callback)));

		} catch (IOException e) {
			throw new HadoopServiceException("error executing hadoop job: " + jobName, e);
		}
	}

	private JobConf initAdminJob(final BlackboardJob bbJob, final String jobName, final JobProfile jobProfile, final JobConf jobConf) throws IOException,
			HadoopServiceException {
		switch (AdminJobType.valueOf(jobName)) {
		case copytable:
			return copyTable(jobProfile, bbJob.getParameters(), jobConf);

		default:
			throw new HadoopServiceException("unknown admin job: " + jobName);
		}
	}

	/**
	 * Builds a CopyTable job. Mimics
	 * <p/>
	 * <pre>
	 * {@code
	 * bin/hbase org.apache.hadoop.hbase.mapreduce.CopyTable --starttime=1265875194289 --endtime=1265878794289
	 * --peer.adr=server1,server2,server3:2181:/hbase --families=myOldCf:myNewCf,cf2,cf3 TestTable <code>
	 *
	 * Options:
	 * rs.class     hbase.regionserver.class of the peer cluster,
	 *              specify if different from current cluster
	 * rs.impl      hbase.regionserver.impl of the peer cluster,
	 * startrow     the start row
	 * stoprow      the stop row
	 * starttime    beginning of the time range (unixtime in millis)
	 *              without endtime means from starttime to forever
	 * endtime      end of the time range.  Ignored if no starttime specified.
	 * versions     number of cell versions to copy
	 * new.name     new table's name
	 * peer.adr     Address of the peer cluster given in the format
	 *              hbase.zookeeer.quorum:hbase.zookeeper.client.port:zookeeper.znode.parent
	 * families     comma-separated list of families to copy
	 *              To copy from cf1 to cf2, give sourceCfName:destCfName.
	 *              To keep the same name, just give "cfName"
	 * all.cells    also copy deleteFromHBase markers and deleted cells
	 *
	 * Args:
	 * tablename    Name of the table to copy
	 * }
	 * </pre>
	 *
	 * @param jobProfile the job profile
	 * @param parameters the parameters
	 * @param jobConf    the job conf
	 * @return the job conf
	 * @throws IOException            Signals that an I/O exception has occurred.
	 * @throws HadoopServiceException
	 */
	private JobConf copyTable(final JobProfile jobProfile, final Map<String, String> parameters, final JobConf jobConf) throws IOException,
			HadoopServiceException {

		ScanProperties scan = jobProfile.getScanProperties();

		List<String> params = Lists.newArrayList();
		if (!scan.getFamilies().isEmpty()) {
			String families = "--families=" + Joiner.on(",").join(scan.getFamilies());
			log.debug("adding column families: " + families);
			params.add(families);
		}

		// copy on remote cluster?
		if (parameters.containsKey("peer.adr") && !StringUtils.equals(parameters.get("sourceCluster"), parameters.get("cluster"))) {
			String peerAdr = "--peer.adr=" + parameters.get("peer.adr");
			log.debug("adding peer address: " + peerAdr);
			params.add(peerAdr);
		}

		// sets the target table
		String targetTable = "--new.name=" + parameters.get("new.name");
		log.debug("adding target table: " + targetTable);
		params.add(targetTable);

		// sets the source table
		String sourceTable = parameters.get("hbase.mapreduce.inputtable");
		log.debug("adding source table: " + sourceTable);
		params.add(sourceTable);

		log.info("copy table params: " + params);

		final Job copyJob = CopyTable.createSubmittableJob(jobConf, Iterables.toArray(params, String.class));

		if (copyJob == null) throw new HadoopServiceException("invalid copytable parameters: " + params);

		final Configuration copyConf = copyJob.getConfiguration();

		return merge(jobConf, copyConf);
	}

	/**
	 * Merge.
	 *
	 * @param jobConf  the job conf
	 * @param copyConf the copy conf
	 * @return the job conf
	 */
	private JobConf merge(final JobConf jobConf, final Configuration copyConf) {
		for (Entry<String, String> e : copyConf) {
			jobConf.set(e.getKey(), e.getValue());
		}
		return jobConf;
	}

}