D-Net

package eu.dnetlib.msro.openaireplus.workflows.nodes.hbase;

import java.util.Map;

import java.util.Set;

import javax.annotation.Resource;

import com.google.common.collect.Sets;

import com.googlecode.sarasvati.Arc;

import com.googlecode.sarasvati.NodeToken;

import eu.dnetlib.data.hadoop.rmi.HadoopService;

import eu.dnetlib.data.hadoop.rmi.HadoopServiceException;

import eu.dnetlib.enabling.locators.UniqueServiceLocator;

import eu.dnetlib.msro.rmi.MSROException;

import eu.dnetlib.msro.workflows.nodes.SimpleJobNode;

import org.apache.commons.lang.StringUtils;

import org.apache.commons.logging.Log;

import org.apache.commons.logging.LogFactory;

/**

 * The PrepareCopyTableJobNode prepares the parameters needed to run the CopyTable job.

 */

public class PrepareCopyTableJobNode extends SimpleJobNode {

	/**

	 * The Constant ZOOKEEPER_ZNODE_PARENT.

	 */

	private static final String ZOOKEEPER_ZNODE_PARENT = "zookeeper.znode.parent";

	/**

	 * The Constant HBASE_ZOOKEEPER_CLIENT_PORT.

	 */

	private static final String HBASE_ZOOKEEPER_CLIENT_PORT = "hbase.zookeeper.client.port";

	/**

	 * The Constant HBASE_ZOOKEEPER_QUORUM.

	 */

	private static final String HBASE_ZOOKEEPER_QUORUM = "hbase.zookeeper.quorum";

	/**

	 * logger.

	 */

	private static final Log log = LogFactory.getLog(PrepareCopyTableJobNode.class);

	/**

	 * The source table.

	 */

	private String sourceCluster;

	/**

	 * The target cluster.

	 */

	private String targetCluster;

	private String sourceTable;

	private String targetTable;

	@Resource

	private UniqueServiceLocator serviceLocator;

	/*

	 * (non-Javadoc)

	 * 

	 * @see eu.dnetlib.msro.workflows.nodes.SimpleJobNode#execute(com.googlecode.sarasvati.NodeToken)

	 */

	@Override

	protected String execute(final NodeToken token) throws Exception {

		checkNodeParams();

		final String outputQuorum = getOutputQuorum();

		log.info("build hbase quorum: " + outputQuorum);

		token.getEnv().setAttribute("peer.adr", outputQuorum);

		token.getEnv().setAttribute("sourceCluster", getSourceCluster());

		token.getEnv().setAttribute("sourceTable", getSourceTable());

		token.getEnv().setAttribute("targetCluster", getTargetCluster());

		token.getEnv().setAttribute("targetTable", getTargetTable());

		return Arc.DEFAULT_ARC;

	}

	/**

	 * Builds the output quorum.

	 *

	 * @return the output quorum

	 * @throws HadoopServiceException when cannot retrieve the clustr configuration

	 * @throws MSROException          when some of the needed properties is missing in the cluster configuration

	 */

	private String getOutputQuorum() throws HadoopServiceException, MSROException {

		Map<String, String> conf = serviceLocator.getService(HadoopService.class).getClusterConfiguration(getTargetCluster());

		log.debug(conf);

		String hbaseQuorum = conf.get(HBASE_ZOOKEEPER_QUORUM);

		String hbasePort = conf.get(HBASE_ZOOKEEPER_CLIENT_PORT);

		String znodeParent = conf.get(ZOOKEEPER_ZNODE_PARENT);

		checkParamExist(hbaseQuorum, String.format("unable to find property '%s' in cluster configuration: %s", HBASE_ZOOKEEPER_QUORUM, hbaseQuorum));

		checkParamExist(hbasePort, String.format("unable to find property '%s' in cluster configuration: %s", HBASE_ZOOKEEPER_CLIENT_PORT, hbasePort));

		checkParamExist(znodeParent, String.format("unable to find property '%s' in cluster configuration: %s", ZOOKEEPER_ZNODE_PARENT, znodeParent));

		String outputQuorum = String.format("%s:%s:%s", hbaseQuorum, hbasePort, znodeParent);

		return outputQuorum;

	}

	/**

	 * Checks the wf params.

	 *

	 * @throws MSROException          the MSRO exception

	 * @throws HadoopServiceException

	 */

	private void checkNodeParams() throws MSROException, HadoopServiceException {

		checkParamExist(getSourceCluster(), "source cluster must be set");

		checkParamExist(getTargetCluster(), "target cluster must be set");

		checkParamExist(getSourceTable(), "source table must be set");

		checkParamExist(getTargetTable(), "target table must be set");

		final HadoopService hadoop = serviceLocator.getService(HadoopService.class);

		Set<String> clusters = Sets.newHashSet(hadoop.listClusters());

		if (!clusters.contains(getSourceCluster())) { throw new MSROException(String.format("source cluster '%s' doesn not exists", getSourceCluster())); }

		if (!clusters.contains(getTargetCluster())) { throw new MSROException(String.format("target cluster '%s' doesn not exists", getTargetCluster())); }

		if (!hadoop.existHbaseTable(getSourceCluster(), getSourceTable())) {

			throw new MSROException(String.format(

					"source table '%s' doesn not exists on cluster '%s'", getSourceTable(), getSourceCluster()));

		}

	}

	/**

	 * Check parameter existence.

	 *

	 * @param param the param

	 * @param msg   the msg

	 * @throws MSROException the MSRO exception

	 */

	private void checkParamExist(final String param, final String msg) throws MSROException {

		if (StringUtils.isBlank(param)) { throw new MSROException(msg); }

	}

	public String getSourceCluster() {

		return sourceCluster;

	}

	public void setSourceCluster(final String sourceCluster) {

		this.sourceCluster = sourceCluster;

	}

	public String getTargetCluster() {

		return targetCluster;

	}

	public void setTargetCluster(final String targetCluster) {

		this.targetCluster = targetCluster;

	}

	public String getSourceTable() {

		return sourceTable;

	}

	public void setSourceTable(final String sourceTable) {

		this.sourceTable = sourceTable;

	}

	public String getTargetTable() {

		return targetTable;

	}

	public void setTargetTable(final String targetTable) {

		this.targetTable = targetTable;

	}

}