D-Net

package eu.dnetlib.data.hadoop;

import java.util.Map;

import java.util.concurrent.ExecutorService;

import java.util.concurrent.Executors;

import java.util.concurrent.TimeUnit;

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

import org.apache.commons.logging.Log;

import org.apache.commons.logging.LogFactory;

import org.apache.hadoop.hbase.client.HBaseAdmin;

import org.apache.hadoop.mapred.JobClient;

import org.apache.oozie.client.OozieClient;

import org.springframework.beans.factory.annotation.Autowired;

import org.springframework.beans.factory.annotation.Required;

import org.springframework.context.annotation.Lazy;

import com.google.common.collect.Maps;

import com.google.gson.Gson;

import eu.dnetlib.data.hadoop.hbase.HBaseAdminFactory;

import eu.dnetlib.data.hadoop.oozie.OozieClientFactory;

import eu.dnetlib.data.hadoop.mapreduce.JobClientFactory;

public class HadoopClientMap {

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

	@Lazy

	@Autowired

	private JobClientFactory jobClientFactory;

	@Lazy

	@Autowired

	private OozieClientFactory oozieClientFactory;

	@Lazy

	@Autowired

	private HBaseAdminFactory hbaseAdminFactory;

	private int clientsInitTime;

	private Map<String, Map<String, String>> enabledClients = Maps.newHashMap();

	private final Map<ClusterName, HadoopClients> clients = Maps.newHashMap();

	private final ExecutorService executor = Executors.newSingleThreadExecutor();

	public void init() {

		log.info("clients conf: " + getEnabledClients());

		executor.execute(new Runnable() {

			@Override

			public void run() {

				for (final String name : enabledClients.keySet()) {

					doInit(name);

				}

			}

		});

	}

	private void doInit(final String name) {

		try {

			log.info("initializing clients for hadoop cluster: " + name);

			final ClusterName clusterName = ClusterName.valueOf(name);

			final Map<String, String> clientsConf = enabledClients.get(name);

			final JobClient jobClient = Boolean.valueOf(clientsConf.get("mapred")) ? getJobClientFactory().newInstance(name) : null;

			final OozieClient oozieClient = Boolean.valueOf(clientsConf.get("oozie")) ? getOozieClientFactory().newInstance(clusterName) : null;

			final HBaseAdmin hbaseAdmin = Boolean.valueOf(clientsConf.get("hbase")) ? getHbaseAdminFactory().newInstance(clusterName) : null;

			clients.put(clusterName, new HadoopClients(jobClient, oozieClient, hbaseAdmin));

		} catch (final Throwable e) {

			log.error("Error initializing hadoop client for cluster: " + name, e);

			throw new RuntimeException(e);

		}

	}

	public boolean waitClients() throws InterruptedException {

		return executor.awaitTermination(getClientsInitTime(), TimeUnit.SECONDS);

	}

	public JobClient getJtClient(final ClusterName name) {

		return getClients(name).getJtClient();

	}

	public boolean isMapreduceAvailable(final ClusterName name) {

		return getClients(name).isMapredAvailable();

	}

	public OozieClient getOozieClient(final ClusterName name) {

		return getClients(name).getOozieClient();

	}

	public boolean isOozieAvailable(final ClusterName name) {

		return getClients(name).isOozieAvailable();

	}

	public HBaseAdmin getHbaseAdmin(final ClusterName name) {

		return getClients(name).getHbaseAdmin();

	}

	public HadoopClients getClients(final ClusterName name) {

		final HadoopClients hadoopClients = clients.get(name);

		if (hadoopClients == null) throw new IllegalArgumentException("cluster " + name.toString() + " is currently disabled");

		return hadoopClients;

	}

	// //////////

	public String getEnabledClients() {

		return new Gson().toJson(enabledClients);

	}

	@Required

	@SuppressWarnings("unchecked")

	public void setEnabledClients(final String enabledClients) {

		this.enabledClients = new Gson().fromJson(enabledClients, Map.class);

	}

	public JobClientFactory getJobClientFactory() {

		return jobClientFactory;

	}

	public void setJobClientFactory(JobClientFactory jobClientFactory) {

		this.jobClientFactory = jobClientFactory;

	}

	public OozieClientFactory getOozieClientFactory() {

		return oozieClientFactory;

	}

	public void setOozieClientFactory(OozieClientFactory oozieClientFactory) {

		this.oozieClientFactory = oozieClientFactory;

	}

	public HBaseAdminFactory getHbaseAdminFactory() {

		return hbaseAdminFactory;

	}

	public void setHbaseAdminFactory(HBaseAdminFactory hbaseAdminFactory) {

		this.hbaseAdminFactory = hbaseAdminFactory;

	}

	public int getClientsInitTime() {

		return clientsInitTime;

	}

	@Required

	public void setClientsInitTime(int clientsInitTime) {

		this.clientsInitTime = clientsInitTime;

	}

}