Index: modules/dnet-hadoop-service/tags/dnet-hadoop-service-2.7.8/pom.xml
===================================================================
--- modules/dnet-hadoop-service/tags/dnet-hadoop-service-2.7.8/pom.xml	(nonexistent)
+++ modules/dnet-hadoop-service/tags/dnet-hadoop-service-2.7.8/pom.xml	(revision 59871)
@@ -0,0 +1,203 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/maven-v4_0_0.xsd">
+	<parent>
+		<groupId>eu.dnetlib</groupId>
+		<artifactId>dnet45-parent</artifactId>
+		<version>1.0.0</version>
+	</parent>
+	<modelVersion>4.0.0</modelVersion>
+	<groupId>eu.dnetlib</groupId>
+	<artifactId>dnet-hadoop-service</artifactId>
+	<packaging>jar</packaging>
+	<version>2.7.8</version>
+	<scm>
+	    <developerConnection>scm:svn:https://svn.driver.research-infrastructures.eu/driver/dnet45/modules/dnet-hadoop-service/tags/dnet-hadoop-service-2.7.8</developerConnection>
+	</scm>
+	<repositories>
+		<!-- Cloudera Repositories -->
+		<repository>
+			<snapshots>
+				<enabled>false</enabled>
+			</snapshots>
+			<id>cloudera-central</id>
+			<name>cloundera-libs-release</name>
+			<url>https://maven.d4science.org/nexus/content/repositories/cloudera-central</url>
+		</repository>
+		<repository>
+			<id>cloudera-snapshots</id>
+			<name>cloudera-libs-snapshot</name>
+			<url>https://maven.d4science.org/nexus/content/repositories/cloudera-snapshots</url>
+		</repository>
+	</repositories>
+	<dependencies>
+		<dependency>
+			<groupId>eu.dnetlib</groupId>
+			<artifactId>dnet-hadoop-service-rmi</artifactId>
+			<version>[1.0.0,2.0.0)</version>
+		</dependency>
+		<dependency>
+			<groupId>com.google.code.gson</groupId>
+			<artifactId>gson</artifactId>
+			<version>${google.gson.version}</version>
+		</dependency>
+		<dependency>
+			<groupId>eu.dnetlib</groupId>
+			<artifactId>cnr-blackboard-common</artifactId>
+			<version>[2.1.0,3.0.0)</version>
+		</dependency>
+		<dependency>
+			<groupId>eu.dnetlib</groupId>
+			<artifactId>cnr-resultset-client</artifactId>
+			<version>[2.0.0,3.0.0)</version>
+			<exclusions>
+				<exclusion>
+					<artifactId>slf4j-api</artifactId>
+					<groupId>org.slf4j</groupId>
+				</exclusion>
+			</exclusions>
+		</dependency>
+		<dependency>
+			<groupId>eu.dnetlib</groupId>
+			<artifactId>dnet-hadoop-commons</artifactId>
+			<version>[2.0.0,3.0.0)</version>
+		</dependency>
+<!--
+		<dependency>
+			<groupId>eu.dnetlib</groupId>
+			<artifactId>dnet-mapreduce-submitter</artifactId>
+			<version>[3.0.0,4.0.0)</version>
+		</dependency>
+-->
+		<dependency>
+			<groupId>org.apache.hbase</groupId>
+			<artifactId>hbase</artifactId>
+			<version>${apache.hbase.version}</version>
+			<exclusions>
+                                <exclusion>
+                                        <artifactId>jasper-compiler</artifactId>
+                                        <groupId>tomcat</groupId>
+                                </exclusion>
+                                <exclusion>
+                                        <artifactId>jasper-runtime</artifactId>
+                                        <groupId>tomcat</groupId>
+                                </exclusion>
+                                <exclusion>
+                                        <artifactId>jamon-runtime</artifactId>
+                                        <groupId>org.jamon</groupId>
+                                </exclusion>
+				<exclusion>
+					<artifactId>jsp-api-2.1</artifactId>
+					<groupId>org.mortbay.jetty</groupId>
+				</exclusion>
+                                <exclusion>
+                                        <artifactId>jsp-2.1</artifactId>
+                                        <groupId>org.mortbay.jetty</groupId>
+                                </exclusion>
+				<exclusion>
+					<artifactId>servlet-api-2.5</artifactId>
+					<groupId>org.mortbay.jetty</groupId>
+				</exclusion>
+                                <exclusion>
+                                        <artifactId>jetty</artifactId>
+                                        <groupId>org.mortbay.jetty</groupId>
+                                </exclusion>
+                                <exclusion>
+                                        <artifactId>jetty-util</artifactId>
+                                        <groupId>org.mortbay.jetty</groupId>
+                                </exclusion>
+				<exclusion>
+					<artifactId>slf4j-api</artifactId>
+					<groupId>org.slf4j</groupId>
+				</exclusion>
+				<exclusion>
+					<artifactId>slf4j-log4j12</artifactId>
+					<groupId>org.slf4j</groupId>
+				</exclusion>
+				<exclusion>
+					<artifactId>stax-api</artifactId>
+					<groupId>stax</groupId>
+				</exclusion>
+				<exclusion>
+					<artifactId>httpclient</artifactId>
+					<groupId>org.apache.httpcomponents</groupId>
+				</exclusion>
+				<exclusion>
+					<artifactId>guava</artifactId>
+					<groupId>com.google.guava</groupId>
+				</exclusion>
+				<exclusion>
+					<artifactId>asm</artifactId>
+					<groupId>asm</groupId>
+				</exclusion>
+				<exclusion>
+					<artifactId>jaxb-api</artifactId>
+					<groupId>javax.xml.bind</groupId>
+				</exclusion>
+				<exclusion>
+					<artifactId>jaxb-impl</artifactId>
+					<groupId>com.sun.xml.bind</groupId>
+				</exclusion>
+				<exclusion>
+					<artifactId>commons-codec</artifactId>
+					<groupId>commons-codec</groupId>
+				</exclusion>
+                                <exclusion>
+                                        <artifactId>commons-logging</artifactId>
+                                        <groupId>commons-logging</groupId>
+                                </exclusion>
+			</exclusions>
+		</dependency>
+		<dependency>
+			<groupId>org.apache.oozie</groupId>
+			<artifactId>oozie-client</artifactId>
+			<version>${apache.oozie.version}</version>
+			<exclusions>
+				<exclusion>
+					<artifactId>slf4j-simple</artifactId>
+					<groupId>org.slf4j</groupId>
+				</exclusion>
+				<exclusion>
+					<artifactId>slf4j-api</artifactId>
+					<groupId>org.slf4j</groupId>
+				</exclusion>
+				<exclusion>
+					<artifactId>xercesImpl</artifactId>
+					<groupId>xerces</groupId>
+				</exclusion>
+			</exclusions>
+		</dependency>
+		<dependency>
+			<groupId>junit</groupId>
+			<artifactId>junit</artifactId>
+			<version>${junit.version}</version>
+			<scope>test</scope>
+		</dependency>
+		<dependency>
+			<groupId>org.apache.logging.log4j</groupId>
+			<artifactId>log4j-slf4j-impl</artifactId>
+			<version>2.0.2</version>
+			<scope>test</scope>
+		</dependency>
+		<dependency>
+			<groupId>org.apache.httpcomponents</groupId>
+			<artifactId>httpclient</artifactId>
+			<version>4.3.1</version>
+		</dependency>
+                <dependency>
+                        <groupId>commons-codec</groupId>
+                        <artifactId>commons-codec</artifactId>
+                        <version>${commons.codec.version}</version>
+                </dependency>
+                <dependency>
+                        <groupId>commons-net</groupId>
+                        <artifactId>commons-net</artifactId>
+                        <version>${commons.net.version}</version>
+                </dependency>
+		<dependency>
+                  <groupId>org.apache.commons</groupId>
+                  <artifactId>commons-compress</artifactId>
+                  <version>${commons.compress.version}</version>
+		</dependency>
+
+	</dependencies>
+</project>
Index: modules/dnet-hadoop-service/tags/dnet-hadoop-service-2.7.8/src/main/java/eu/dnetlib/data/hadoop/action/ImportEprHdfsAction.java
===================================================================
--- modules/dnet-hadoop-service/tags/dnet-hadoop-service-2.7.8/src/main/java/eu/dnetlib/data/hadoop/action/ImportEprHdfsAction.java	(nonexistent)
+++ modules/dnet-hadoop-service/tags/dnet-hadoop-service-2.7.8/src/main/java/eu/dnetlib/data/hadoop/action/ImportEprHdfsAction.java	(revision 59871)
@@ -0,0 +1,53 @@
+package eu.dnetlib.data.hadoop.action;
+
+import java.io.IOException;
+
+import org.apache.commons.lang.StringUtils;
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.springframework.beans.factory.annotation.Required;
+
+import eu.dnetlib.data.hadoop.config.ClusterName;
+import eu.dnetlib.data.hadoop.rmi.HadoopServiceException;
+import eu.dnetlib.enabling.tools.blackboard.BlackboardJob;
+import eu.dnetlib.enabling.tools.blackboard.BlackboardServerHandler;
+import eu.dnetlib.miscutils.functional.xml.DnetXsltFunctions;
+
+public class ImportEprHdfsAction extends AbstractHadoopAction {
+
+	private static final Log log = LogFactory.getLog(ImportEprHdfsAction.class); // NOPMD by marko on 11/24/08 5:02 PM
+
+	private SequenceFileFeeder sequenceFileFeeder;
+
+	@Override
+	public void executeAsync(final BlackboardServerHandler handler, final BlackboardJob job) throws HadoopServiceException {
+
+		final String epr = DnetXsltFunctions.decodeBase64(job.getParameters().get("input_epr"));
+		final ClusterName clusterName = ClusterName.valueOf(job.getParameters().get("cluster"));
+		final String path = job.getParameters().get("path");
+
+		log.info(String.format("Starting import in hdfs sequence file '%s', cluster '%s', epr '%s...", path, clusterName.toString(),
+				StringUtils.substring(epr, 0, 20)));
+		try {
+			Integer count = getSequenceFileFeeder().feed(epr, clusterName, path);
+			if (count != null) {
+				log.info("Import completed successfully");
+				job.getParameters().put("count", String.valueOf(count));
+			}
+
+			handler.done(job);
+		} catch (IOException e) {
+			throw new HadoopServiceException("Import failed", e);
+		}
+	}
+
+	public SequenceFileFeeder getSequenceFileFeeder() {
+		return sequenceFileFeeder;
+	}
+
+	@Required
+	public void setSequenceFileFeeder(final SequenceFileFeeder sequenceFileFeeder) {
+		this.sequenceFileFeeder = sequenceFileFeeder;
+	}
+
+}
Index: modules/dnet-hadoop-service/tags/dnet-hadoop-service-2.7.8/src/main/java/eu/dnetlib/data/hadoop/HadoopServiceImpl.java
===================================================================
--- modules/dnet-hadoop-service/tags/dnet-hadoop-service-2.7.8/src/main/java/eu/dnetlib/data/hadoop/HadoopServiceImpl.java	(nonexistent)
+++ modules/dnet-hadoop-service/tags/dnet-hadoop-service-2.7.8/src/main/java/eu/dnetlib/data/hadoop/HadoopServiceImpl.java	(revision 59871)
@@ -0,0 +1,346 @@
+package eu.dnetlib.data.hadoop;
+
+import java.io.IOException;
+import java.util.List;
+import java.util.Map;
+import java.util.Map.Entry;
+import java.util.Set;
+
+import com.google.common.collect.Lists;
+import com.google.common.collect.Maps;
+import eu.dnetlib.data.hadoop.config.ClusterName;
+import eu.dnetlib.data.hadoop.rmi.HadoopJobDescriptor;
+import eu.dnetlib.data.hadoop.rmi.HadoopService;
+import eu.dnetlib.data.hadoop.rmi.HadoopServiceException;
+import eu.dnetlib.data.hadoop.rmi.hbase.HBaseRowDescriptor;
+import eu.dnetlib.enabling.tools.AbstractBaseService;
+import eu.dnetlib.enabling.tools.blackboard.NotificationHandler;
+import org.apache.commons.codec.binary.Base64;
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.hbase.client.Put;
+import org.apache.hadoop.hbase.util.Bytes;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.beans.factory.annotation.Required;
+
+/**
+ * The Class HadoopServiceImpl.
+ */
+public class HadoopServiceImpl extends AbstractBaseService implements HadoopService {
+
+	private static final Log log = LogFactory.getLog(HadoopServiceImpl.class);
+	/**
+	 * notification handler.
+	 */
+	private NotificationHandler notificationHandler;
+
+	/** The hadoop service core. */
+	@Autowired
+	private HadoopServiceCore hadoopServiceCore;
+
+	/** The job registry. */
+	@Autowired
+	private JobRegistry jobRegistry;
+
+	/*
+	 * (non-Javadoc)
+	 * 
+	 * @see eu.dnetlib.data.hadoop.rmi.HadoopService#listAvailableJobs()
+	 */
+	@Override
+	public List<String> listAvailableJobs() throws HadoopServiceException {
+		List<String> res = Lists.newArrayList();
+		return res;
+	}
+
+	/*
+	 * (non-Javadoc)
+	 * 
+	 * @see eu.dnetlib.data.hadoop.rmi.HadoopService#listJobs(java.lang.String)
+	 */
+	@Override
+	public List<HadoopJobDescriptor> listJobs(final String clusterName) throws HadoopServiceException {
+		return jobRegistry.listJobs(checkExists(clusterName));
+	}
+
+	/*
+	 * (non-Javadoc)
+	 * 
+	 * @see eu.dnetlib.data.hadoop.rmi.HadoopService#killJob(java.lang.String)
+	 */
+	@Override
+	public boolean killJob(final String jobId) throws HadoopServiceException {
+		jobRegistry.unregisterJob(jobId);
+		return true;
+	}
+
+	/*
+	 * (non-Javadoc)
+	 * 
+	 * @see eu.dnetlib.enabling.tools.AbstractBaseService#notify(java.lang.String, java.lang.String, java.lang.String, java.lang.String)
+	 */
+	@Override
+	public void notify(final String subscriptionId, final String topic, final String isId, final String message) {
+		getNotificationHandler().notified(subscriptionId, topic, isId, message);
+	}
+
+	/*
+	 * (non-Javadoc)
+	 * 
+	 * @see eu.dnetlib.data.hadoop.rmi.HadoopService#listHbaseTables(java.lang.String)
+	 */
+	@Override
+	public List<String> listHbaseTables(final String clusterName) throws HadoopServiceException {
+		try {
+			return hadoopServiceCore.listTables(checkExists(clusterName));
+		} catch (IOException e) {
+			throw new HadoopServiceException(e);
+		}
+	}
+
+	/*
+	 * (non-Javadoc)
+	 * 
+	 * @see eu.dnetlib.data.hadoop.rmi.HadoopService#ensureHbaseTable(java.lang.String, java.lang.String, java.util.Set)
+	 */
+	@Override
+	public boolean ensureHbaseTable(final String clusterName, final String tableName, final Set<String> columns) throws HadoopServiceException {
+		try {
+			hadoopServiceCore.ensureTable(checkExists(clusterName), tableName, columns);
+			return true;
+		} catch (IOException e) {
+			throw new HadoopServiceException(e);
+		}
+	}
+
+	/*
+	 * (non-Javadoc)
+	 * 
+	 * @see eu.dnetlib.data.hadoop.rmi.HadoopService#createHbaseTable(java.lang.String, java.lang.String, java.util.Set)
+	 */
+	@Override
+	public boolean createHbaseTable(final String clusterName, final String tableName, final Set<String> columns) throws HadoopServiceException {
+		try {
+			hadoopServiceCore.createTable(checkExists(clusterName), tableName, columns);
+			return true;
+		} catch (IOException e) {
+			throw new HadoopServiceException(e);
+		}
+	}
+
+	/*
+	 * (non-Javadoc)
+	 *
+	 * @see eu.dnetlib.data.hadoop.rmi.HadoopService#createHbaseTable(java.lang.String, java.lang.String, java.util.String)
+	 */
+	@Override
+	public boolean createConfiguredHbaseTable(final String clusterName, final String tableName, final String tableConfiguration) throws HadoopServiceException {
+		try {
+			hadoopServiceCore.createTable(checkExists(clusterName), tableName, tableConfiguration);
+			return true;
+		} catch (IOException e) {
+			throw new HadoopServiceException(e);
+		}
+	}
+
+	/*
+	 * (non-Javadoc)
+	 * 
+	 * @see eu.dnetlib.data.hadoop.rmi.HadoopService#truncateHbaseTable(java.lang.String, java.lang.String)
+	 */
+	@Override
+	public boolean truncateHbaseTable(final String clusterName, final String tableName) throws HadoopServiceException {
+		try {
+			hadoopServiceCore.truncateTable(checkExists(clusterName), tableName);
+		} catch (IOException e) {
+			throw new HadoopServiceException(e);
+		}
+		return true;
+	}
+
+	/*
+	 * (non-Javadoc)
+	 * 
+	 * @see eu.dnetlib.data.hadoop.rmi.HadoopService#dropHbaseTable(java.lang.String, java.lang.String)
+	 */
+	@Override
+	public boolean dropHbaseTable(final String clusterName, final String tableName) throws HadoopServiceException {
+		try {
+			hadoopServiceCore.dropTable(checkExists(clusterName), tableName);
+		} catch (IOException e) {
+			throw new HadoopServiceException(e);
+		}
+		return true;
+	}
+
+	/*
+	 * (non-Javadoc)
+	 * 
+	 * @see eu.dnetlib.data.hadoop.rmi.HadoopService#existHbaseTable(java.lang.String, java.lang.String)
+	 */
+	@Override
+	public boolean existHbaseTable(final String clusterName, final String tableName) throws HadoopServiceException {
+		try {
+			return hadoopServiceCore.existTable(checkExists(clusterName), tableName);
+		} catch (IOException e) {
+			throw new HadoopServiceException(e);
+		}
+	}
+
+	/*
+	 * (non-Javadoc)
+	 * 
+	 * @see eu.dnetlib.data.hadoop.rmi.HadoopService#getClusterConfiguration(java.lang.String)
+	 */
+	@Override
+	public Map<String, String> getClusterConfiguration(final String clusterName) throws HadoopServiceException {
+
+		final Configuration conf = hadoopServiceCore.getClusterConiguration(checkExists(clusterName));
+		final Map<String, String> res = Maps.newHashMap();
+		for (Entry<String, String> e : conf) {
+			res.put(e.getKey(), e.getValue());
+		}
+
+		return res;
+	}
+
+	@Override
+	public boolean deleteHdfsPath(final String clusterName, final String path) throws HadoopServiceException {
+		return hadoopServiceCore.deleteFromHdfs(checkExists(clusterName), path);
+	}
+
+	@Override
+	public boolean existHdfsPath(final String clusterName, final String path) throws HadoopServiceException {
+		return hadoopServiceCore.existHdfsPath(checkExists(clusterName), path);
+	}
+
+	@Override
+	public boolean createHdfsDirectory(final String clusterName, final String path, final boolean force) throws HadoopServiceException {
+		return hadoopServiceCore.createHdfsDir(checkExists(clusterName), path, force);
+	}
+
+	@Override
+	public List<String> listClusters() throws HadoopServiceException {
+		try {
+			return ClusterName.asStringList();
+		} catch (Throwable e) {
+			throw new HadoopServiceException(e);
+		}
+	}
+
+	@Override
+	public List<String> describeHbaseTable(final String clusterName, final String tableName) throws HadoopServiceException {
+		try {
+			return hadoopServiceCore.describeTable(checkExists(clusterName), tableName);
+		} catch (IOException e) {
+			throw new HadoopServiceException(e);
+		}
+	}
+
+	@Override
+	public HBaseRowDescriptor describeHBaseColumn(final String clusterName, final String tableName, final String rowKey) throws HadoopServiceException {
+		try {
+			return hadoopServiceCore.describeRow(checkExists(clusterName), tableName, rowKey);
+		} catch (IOException e) {
+			throw new HadoopServiceException(e);
+		}
+	}
+
+	@Override
+	public Map<String, HBaseRowDescriptor> describeHBaseColumns(final String clusterName, final String tableName, final List<String> rowKeys) throws HadoopServiceException {
+		try {
+			return hadoopServiceCore.describeRows(checkExists(clusterName), tableName, rowKeys);
+		} catch (IOException e) {
+			throw new HadoopServiceException(e);
+		}
+	}
+
+	@Override
+	public String describeHBaseTableConfiguration(final String clusterName, final String tableName) throws HadoopServiceException {
+		try {
+			return hadoopServiceCore.getHBaseTableDescriptor(checkExists(clusterName), tableName);
+		} catch (IOException e) {
+			throw new HadoopServiceException(e);
+		}
+	}
+
+	@Override
+	public boolean deleteHBaseColumn(final String clusterName, final String tableName, final HBaseRowDescriptor column) throws HadoopServiceException {
+		try {
+			hadoopServiceCore.deleteColumnsFromHBase(checkExists(clusterName), tableName, Lists.newArrayList(column));
+			return true;
+		} catch (IOException e) {
+			throw new HadoopServiceException(e);
+		}
+	}
+
+	@Override
+	public boolean deleteHBaseColumns(final String clusterName, final String tableName, final List<HBaseRowDescriptor> column) throws HadoopServiceException {
+		try {
+			hadoopServiceCore.deleteColumnsFromHBase(checkExists(clusterName), tableName, column);
+			return true;
+		} catch (IOException e) {
+			throw new HadoopServiceException(e);
+		}
+	}
+
+	@Override
+	public void addHBaseColumn(final String clusterName,
+			final String tableName,
+			final String rowKey,
+			final String columnFamily,
+			final String qualifier,
+			final String value)
+			throws HadoopServiceException {
+		log.debug(String.format("adding value to K: '%s' CF:'%s' Q:'%s'", rowKey, columnFamily, qualifier));
+		final Put put = new Put(Bytes.toBytes(rowKey)).add(Bytes.toBytes(columnFamily), Bytes.toBytes(qualifier), Base64.decodeBase64(value));
+		try {
+			hadoopServiceCore.writeOnHBase(ClusterName.valueOf(clusterName), tableName, Lists.newArrayList(put));
+		} catch (IOException e) {
+			throw new HadoopServiceException(e);
+		}
+	}
+
+
+
+	///////////////////
+
+	/**
+	 * Check exists.
+	 *
+	 * @param clusterName
+	 *            the cluster name
+	 * @return the cluster name
+	 * @throws HadoopServiceException
+	 *             the hadoop service exception
+	 */
+	private ClusterName checkExists(final String clusterName) throws HadoopServiceException {
+		try {
+			return ClusterName.valueOf(clusterName);
+		} catch (final IllegalArgumentException e) {
+			throw new HadoopServiceException("Invalid cluster name: " + clusterName);
+		}
+	}
+
+	/**
+	 * Gets the notification handler.
+	 *
+	 * @return the notification handler
+	 */
+	public NotificationHandler getNotificationHandler() {
+		return notificationHandler;
+	}
+
+	/**
+	 * Sets the notification handler.
+	 *
+	 * @param notificationHandler
+	 *            the new notification handler
+	 */
+	@Required
+	public void setNotificationHandler(final NotificationHandler notificationHandler) {
+		this.notificationHandler = notificationHandler;
+	}
+
+}
Index: modules/dnet-hadoop-service/tags/dnet-hadoop-service-2.7.8/src/main/resources/eu/dnetlib/data/hadoop/applicationContext-dnet-hadoop-service.properties
===================================================================
--- modules/dnet-hadoop-service/tags/dnet-hadoop-service-2.7.8/src/main/resources/eu/dnetlib/data/hadoop/applicationContext-dnet-hadoop-service.properties	(nonexistent)
+++ modules/dnet-hadoop-service/tags/dnet-hadoop-service-2.7.8/src/main/resources/eu/dnetlib/data/hadoop/applicationContext-dnet-hadoop-service.properties	(revision 59871)
@@ -0,0 +1,7 @@
+services.hadoop.clients={"DM":{"oozie":"true","mapred":"true","hbase":"true"},"IIS":{"oozie":"true","mapred":"false","hbase":"false"},"DHP":{"oozie":"true","mapred":"false","hbase":"false"}}
+services.hadoop.hbase.tablefeeder.batchsize=500
+services.hadoop.jobregistry.size=100
+services.hadoop.lib.path=/user/dnet/lib/dnet-mapreduce-jobs-assembly-0.0.6.2-SNAPSHOT.jar
+services.hadoop.hbase.maxversions=1
+services.hadoop.clients.init.timeout=30
+services.hadoop.user=dnet
Index: modules/dnet-hadoop-service/tags/dnet-hadoop-service-2.7.8/src/main/resources/eu/dnetlib/data/hadoop/applicationContext-dnet-hadoop-service.xml
===================================================================
--- modules/dnet-hadoop-service/tags/dnet-hadoop-service-2.7.8/src/main/resources/eu/dnetlib/data/hadoop/applicationContext-dnet-hadoop-service.xml	(nonexistent)
+++ modules/dnet-hadoop-service/tags/dnet-hadoop-service-2.7.8/src/main/resources/eu/dnetlib/data/hadoop/applicationContext-dnet-hadoop-service.xml	(revision 59871)
@@ -0,0 +1,122 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<beans xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+       xmlns:jaxws="http://cxf.apache.org/jaxws" xmlns:p="http://www.springframework.org/schema/p"
+       xmlns:http="http://cxf.apache.org/transports/http/configuration" xmlns:t="http://dnetlib.eu/springbeans/t"
+       xmlns:template="http://dnetlib.eu/springbeans/template" xmlns="http://www.springframework.org/schema/beans"
+       xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd
+
+
+                                    http://cxf.apache.org/transports/http/configuration http://cxf.apache.org/schemas/configuration/http-conf.xsd
+                            http://cxf.apache.org/jaxws http://cxf.apache.org/schemas/jaxws.xsd
+
+                            http://dnetlib.eu/springbeans/template http://dnetlib.eu/springbeans/template.xsd">
+
+
+	<bean id="hadoopService" class="eu.dnetlib.data.hadoop.HadoopServiceImpl"
+		init-method="start" destroy-method="stop" p:notificationHandler-ref="hadoopNotificationHandler" />
+
+	<bean id="hadoopNotificationHandler"
+		class="eu.dnetlib.enabling.tools.blackboard.BlackboardServerExecutorNotificationHandler"
+		p:blackboardExecutor-ref="hadoopBlackboardExecutor" />
+
+	<bean id="hadoopBlackboardExecutor"
+		class="eu.dnetlib.enabling.tools.blackboard.BlackboardServerActionExecutor"
+		p:blackboardHandler-ref="blackboardHandler" p:actionType="eu.dnetlib.data.hadoop.rmi.HadoopBlackboardActions"
+		p:incomplete="false">
+		<property name="actionMap">
+			<map>
+				<entry key="SUBMIT_MAPREDUCE_JOB">
+					<bean class="eu.dnetlib.data.hadoop.action.SubmitMapreduceJobAction" />
+				</entry>
+				<entry key="SUBMIT_ADMIN_JOB">
+					<bean class="eu.dnetlib.data.hadoop.action.SubmitAdminJobAction" />
+				</entry>				
+				<entry key="SUBMIT_OOZIE_JOB">
+					<bean class="eu.dnetlib.data.hadoop.action.SubmitOozieJobAction" />
+				</entry>
+				<entry key="KILL_JOB">
+					<bean class="eu.dnetlib.data.hadoop.action.KillJobAction" />
+				</entry>
+				<entry key="IMPORT_EPR_HBASE">
+					<bean class="eu.dnetlib.data.hadoop.action.ImportEprHbaseAction">
+						<property name="hbaseTableFeeder">
+							<bean class="eu.dnetlib.data.hadoop.hbase.HBasePutFeeder"
+								p:batchSize="${services.hadoop.hbase.tablefeeder.batchsize}" 
+								p:resultSetClientFactory-ref="resultSetClientFactory" />
+						</property>
+					</bean>
+				</entry>
+				<entry key="DELETE_EPR_HBASE">
+					<bean class="eu.dnetlib.data.hadoop.action.ImportEprHbaseAction">
+						<property name="hbaseTableFeeder">
+							<bean class="eu.dnetlib.data.hadoop.hbase.HBaseDeleteFeeder"
+								p:batchSize="${services.hadoop.hbase.tablefeeder.batchsize}" 
+								p:resultSetClientFactory-ref="resultSetClientFactory" />
+						</property>
+					</bean>
+				</entry>
+				<entry key="IMPORT_EPR_HDFS">
+					<bean class="eu.dnetlib.data.hadoop.action.ImportEprHdfsAction">
+						<property name="sequenceFileFeeder">
+							<bean class="eu.dnetlib.data.hadoop.action.SequenceFileFeeder"
+								p:resultSetClientFactory-ref="resultSetClientFactory" />
+						</property>
+					</bean>
+				</entry>
+				<entry key="DELETE_HDFS_PATH">
+					<bean class="eu.dnetlib.data.hadoop.action.DeleteHdfsPathAction" />
+				</entry>
+				<entry key="CREATE_HDFS_DIR">
+					<bean class="eu.dnetlib.data.hadoop.action.CreateHdfsDirectoryAction"/>
+				</entry>
+			</map>
+		</property>
+	</bean>
+
+	<bean id="hadoopServiceCore" class="eu.dnetlib.data.hadoop.HadoopServiceCore" 
+		p:maxVersions="${services.hadoop.hbase.maxversions}"/>
+
+    <bean id="isClient" class="eu.dnetlib.data.hadoop.ISClient" />
+
+	<bean id="hadoopClientMap" class="eu.dnetlib.data.hadoop.HadoopClientMap"
+		p:enabledClients="${services.hadoop.clients}"
+		p:jobClientFactory-ref="jobClientFactory"
+		p:hbaseAdminFactory-ref="hbaseAdminFactory"
+		p:oozieClientFactory-ref="oozieClientFactory" />
+
+	<bean id="oozieClientFactory" class="eu.dnetlib.data.hadoop.oozie.OozieClientFactory" />
+
+	<bean id="hbaseAdminFactory" class="eu.dnetlib.data.hadoop.hbase.HBaseAdminFactory" />
+
+	<bean id="jobClientFactory" class="eu.dnetlib.data.hadoop.mapred.JobClientFactory" />
+		
+	<bean id="hadoopJobRegistry" class="eu.dnetlib.data.hadoop.JobRegistry" 
+		p:maxJobs="${services.hadoop.jobregistry.size}"/>
+
+	<bean id="hadoopServiceLocator" class="eu.dnetlib.enabling.tools.StaticServiceLocator" 
+		p:service-ref="hadoopService" />
+
+	<!-- endpoints -->
+	<jaxws:endpoint id="hadoopServiceEndpoint" implementor="#hadoopService"
+		implementorClass="eu.dnetlib.data.hadoop.rmi.HadoopService" address="/hadoop" />
+
+	<template:instance name="serviceRegistrationManager"
+		t:serviceRegistrationManagerClass="eu.dnetlib.enabling.tools.registration.ValidatingServiceRegistrationManagerImpl"
+		t:name="hadoopServiceRegistrationManager" t:service="hadoopService"
+		t:endpoint="hadoopServiceEndpoint" t:jobScheduler="jobScheduler"
+		t:serviceRegistrator="hadoopServiceRegistrator" />
+		
+	<bean id="hadoopServiceRegistrator"
+		class="eu.dnetlib.enabling.tools.registration.BlackboardServiceRegistrator"
+		p:serviceLocator-ref="uniqueServiceLocator" p:serviceNameResolver-ref="defaultServiceNameResolver"
+		p:hnmLocator-ref="${services.registration.default.hnmlocator}"
+		p:eprBuilder-ref="jaxwsEndpointReferenceBuilder">
+		<property name="serviceProperties">
+			<map>
+				<entry key="defaultLibPath" value="${services.hadoop.lib.path}" />
+				<entry key="default.hadoop.user" value="${services.hadoop.user}" />
+			</map>
+		</property>
+	</bean>		
+
+</beans>
Index: modules/dnet-hadoop-service/tags/dnet-hadoop-service-2.7.8/src/test/java/eu/dnetlib/data/hadoop/utils/ReadSequenceFileTest.java
===================================================================
--- modules/dnet-hadoop-service/tags/dnet-hadoop-service-2.7.8/src/test/java/eu/dnetlib/data/hadoop/utils/ReadSequenceFileTest.java	(nonexistent)
+++ modules/dnet-hadoop-service/tags/dnet-hadoop-service-2.7.8/src/test/java/eu/dnetlib/data/hadoop/utils/ReadSequenceFileTest.java	(revision 59871)
@@ -0,0 +1,74 @@
+package eu.dnetlib.data.hadoop.utils;
+
+import java.util.Map;
+import java.util.Map.Entry;
+
+import org.apache.commons.math.stat.descriptive.SummaryStatistics;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.io.Text;
+import org.junit.Before;
+import org.junit.Ignore;
+import org.junit.Test;
+import org.springframework.core.io.ClassPathResource;
+
+import com.google.common.collect.Maps;
+
+import eu.dnetlib.data.hadoop.config.ConfigurationFactory;
+import eu.dnetlib.data.hadoop.hdfs.SequenceFileUtils;
+import eu.dnetlib.miscutils.collections.Pair;
+
+public class ReadSequenceFileTest {
+
+	private static final Path SEQUENCE_FILE_PATH = new Path("hdfs://nmis-hadoop-cluster/tmp/indexrecords_db_openaireplus_sesam_SESAMALL.seq");
+	private final String HADOOP_CONF_FILE = "/eu/dnetlib/data/hadoop/config/hadoop-default.dm.cnr.properties";
+
+	private Configuration conf;
+
+	@Before
+	public void setUp() {
+		final ConfigurationFactory confFactory = new ConfigurationFactory();
+		confFactory.setDefaults(new ClassPathResource(HADOOP_CONF_FILE));
+		conf = confFactory.getConfiguration();
+	}
+
+	@Test
+	@Ignore
+	public void testReadSequenceFile() throws Exception {
+		final SummaryStatistics statsAll = new SummaryStatistics();
+
+		final Map<String, SummaryStatistics> stats = Maps.newHashMap();
+
+		int i = 0;
+		for (Pair<Text, Text> pair : SequenceFileUtils.read(SEQUENCE_FILE_PATH, conf)) {
+			final String id = pair.getKey().toString();
+			final String record = pair.getValue().toString();
+			final int length = record.getBytes().length;
+
+			final String type = id.substring(0, 2);
+			if (!stats.containsKey(type)) {
+				stats.put(type, new SummaryStatistics());
+			}
+			statsAll.addValue(length);
+			stats.get(type).addValue(length);
+
+			if (++i % 10000 == 0) {
+				System.out.println("Read " + i);
+			}
+		}
+
+		printStats("ALL", statsAll);
+		for (Entry<String, SummaryStatistics> e : stats.entrySet()) {
+			printStats(e.getKey(), e.getValue());
+		}
+	}
+
+	private void printStats(final String type, final SummaryStatistics stats) {
+		System.out.println("************************************");
+		System.out.println("Type: " + type);
+		System.out.println(String.format("\tmin    : %.2f KBytes", stats.getMin() / 1024));
+		System.out.println(String.format("\tmax    : %.2f KBytes", stats.getMax() / 1024));
+		System.out.println(String.format("\tavg    : %.2f KBytes", stats.getMean() / 1024));
+		System.out.println(String.format("\tstdDev : %.2f", stats.getStandardDeviation() / 1024));
+	}
+}
Index: modules/dnet-hadoop-service/tags/dnet-hadoop-service-2.7.8/deploy.info
===================================================================
--- modules/dnet-hadoop-service/tags/dnet-hadoop-service-2.7.8/deploy.info	(nonexistent)
+++ modules/dnet-hadoop-service/tags/dnet-hadoop-service-2.7.8/deploy.info	(revision 59871)
@@ -0,0 +1 @@
+{"type_source": "SVN", "goal": "package -U -T 4C source:jar", "url": "http://svn-public.driver.research-infrastructures.eu/driver/dnet45/modules/dnet-hadoop-service/trunk/", "deploy_repository": "dnet45-snapshots", "version": "4", "mail": "sandro.labruzzo@isti.cnr.it,michele.artini@isti.cnr.it, claudio.atzori@isti.cnr.it, alessia.bardi@isti.cnr.it", "deploy_repository_url": "http://maven.research-infrastructures.eu/nexus/content/repositories/dnet45-snapshots", "name": "dnet-hadoop-service"}
\ No newline at end of file
Index: modules/dnet-hadoop-service/tags/dnet-hadoop-service-2.7.8/src/main/java/eu/dnetlib/data/hadoop/HadoopJob.java
===================================================================
--- modules/dnet-hadoop-service/tags/dnet-hadoop-service-2.7.8/src/main/java/eu/dnetlib/data/hadoop/HadoopJob.java	(nonexistent)
+++ modules/dnet-hadoop-service/tags/dnet-hadoop-service-2.7.8/src/main/java/eu/dnetlib/data/hadoop/HadoopJob.java	(revision 59871)
@@ -0,0 +1,96 @@
+package eu.dnetlib.data.hadoop;
+
+import java.util.Date;
+import java.util.concurrent.Executor;
+import java.util.concurrent.Executors;
+
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+
+import eu.dnetlib.data.hadoop.action.JobMonitor;
+import eu.dnetlib.data.hadoop.config.ClusterName;
+import eu.dnetlib.data.hadoop.rmi.HadoopJobDescriptor;
+import eu.dnetlib.data.hadoop.rmi.HadoopServiceException;
+import eu.dnetlib.data.hadoop.utils.JobProfile;
+
+public class HadoopJob {
+
+	private static final Log log = LogFactory.getLog(HadoopJob.class); // NOPMD by marko on 11/24/08 5:02 PM
+
+	/**
+	 * Defines the possible stati of an hadoop job.
+	 */
+	public static enum Status {
+		PREP, RUNNING, SUCCEEDED, KILLED, FAILED, SUSPENDED, UNKNOWN
+	}
+
+	private final Executor executor = Executors.newSingleThreadExecutor();
+
+	private final JobMonitor jobMonitor;
+
+	private final JobProfile jobProfile;
+
+	private final ClusterName clusterName;
+
+	private final String id;
+
+	public static HadoopJob newInstance(String id, ClusterName clusterName, JobProfile profile, JobMonitor jobMonitor) {
+		return new HadoopJob(id, clusterName, profile, jobMonitor);
+	}
+
+	private HadoopJob(String id, ClusterName clusterName, JobProfile jobProfile, JobMonitor jobMonitor) {
+		super();
+		this.id = id;
+		this.clusterName = clusterName;
+		this.jobProfile = jobProfile;
+		this.jobMonitor = jobMonitor;
+	}
+
+	public void startMonitor() {
+		log.info("start monitoring for job: " + getId());
+		executor.execute(jobMonitor);
+	}
+
+	public String getId() {
+		return id;
+	}
+
+	public String getHadoopId() {
+		return getJobMonitor().getHadoopId();
+	}
+
+	public JobMonitor getJobMonitor() {
+		return jobMonitor;
+	}
+
+	public Status getStatus() {
+		return Status.valueOf(getJobMonitor().getStatus());
+	}
+
+	public boolean isComplete() {
+		Status status = getStatus();
+		return status.equals(Status.SUCCEEDED) || status.equals(Status.FAILED) || status.equals(Status.KILLED);
+	}
+
+	public Date getStartTime() throws HadoopServiceException {
+		return jobMonitor.getStartTime();
+	}
+
+	public Date getLastActivity() {
+		return jobMonitor.getLastActivity();
+	}
+
+	public ClusterName getClusterName() {
+		return clusterName;
+	}
+
+	public JobProfile getJobProfile() {
+		return jobProfile;
+	}
+
+	public HadoopJobDescriptor asDescriptor() throws HadoopServiceException {
+		return new HadoopJobDescriptor(getJobProfile().getName(), getJobProfile().getDescription(), getId(), getStatus().toString(), getStartTime().toString(),
+				getLastActivity().toString(), getJobMonitor().getHadoopId(), getJobMonitor().getTrackerUrl());
+	}
+
+}
Index: modules/dnet-hadoop-service/tags/dnet-hadoop-service-2.7.8/src/main/java/eu/dnetlib/data/hadoop/action/KillJobAction.java
===================================================================
--- modules/dnet-hadoop-service/tags/dnet-hadoop-service-2.7.8/src/main/java/eu/dnetlib/data/hadoop/action/KillJobAction.java	(nonexistent)
+++ modules/dnet-hadoop-service/tags/dnet-hadoop-service-2.7.8/src/main/java/eu/dnetlib/data/hadoop/action/KillJobAction.java	(revision 59871)
@@ -0,0 +1,33 @@
+package eu.dnetlib.data.hadoop.action;
+
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.springframework.beans.factory.annotation.Autowired;
+
+import eu.dnetlib.data.hadoop.JobRegistry;
+import eu.dnetlib.data.hadoop.rmi.HadoopServiceException;
+import eu.dnetlib.enabling.tools.blackboard.BlackboardJob;
+import eu.dnetlib.enabling.tools.blackboard.BlackboardServerHandler;
+
+public class KillJobAction extends AbstractHadoopAction {
+
+	private static final Log log = LogFactory.getLog(KillJobAction.class); // NOPMD by marko on 11/24/08 5:02 PM
+
+	@Autowired
+	private JobRegistry jobRegistry;
+
+	@Override
+	public void executeAsync(final BlackboardServerHandler handler, final BlackboardJob bbJob) {
+
+		final String jobid = bbJob.getParameters().get("job.id");
+		log.info("killing job: " + jobid);
+
+		try {
+			jobRegistry.unregisterJob(jobid);
+			handler.done(bbJob);
+		} catch (HadoopServiceException e) {
+			handler.failed(bbJob, e);
+		}
+	}
+
+}
Index: modules/dnet-hadoop-service/tags/dnet-hadoop-service-2.7.8/src/main/java/eu/dnetlib/data/hadoop/utils/HadoopUtils.java
===================================================================
--- modules/dnet-hadoop-service/tags/dnet-hadoop-service-2.7.8/src/main/java/eu/dnetlib/data/hadoop/utils/HadoopUtils.java	(nonexistent)
+++ modules/dnet-hadoop-service/tags/dnet-hadoop-service-2.7.8/src/main/java/eu/dnetlib/data/hadoop/utils/HadoopUtils.java	(revision 59871)
@@ -0,0 +1,24 @@
+package eu.dnetlib.data.hadoop.utils;
+
+import eu.dnetlib.data.hadoop.HadoopJob;
+import eu.dnetlib.data.hadoop.rmi.HadoopJobDescriptor;
+import eu.dnetlib.data.hadoop.rmi.HadoopServiceException;
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+
+public class HadoopUtils {
+
+	private static final Log log = LogFactory.getLog(HadoopUtils.class);
+
+	public static java.util.function.Function<HadoopJob, HadoopJobDescriptor> asDescriptor() {
+		return d -> {
+			try {
+				return d.asDescriptor();
+			} catch (HadoopServiceException e) {
+				log.warn(e);
+				return null;
+			}
+		};
+	}
+
+}
Index: modules/dnet-hadoop-service/tags/dnet-hadoop-service-2.7.8/src/main/java/eu/dnetlib/data/hadoop/oozie/OozieJobMonitor.java
===================================================================
--- modules/dnet-hadoop-service/tags/dnet-hadoop-service-2.7.8/src/main/java/eu/dnetlib/data/hadoop/oozie/OozieJobMonitor.java	(nonexistent)
+++ modules/dnet-hadoop-service/tags/dnet-hadoop-service-2.7.8/src/main/java/eu/dnetlib/data/hadoop/oozie/OozieJobMonitor.java	(revision 59871)
@@ -0,0 +1,175 @@
+package eu.dnetlib.data.hadoop.oozie;
+
+import java.io.IOException;
+import java.util.*;
+
+import com.google.common.collect.Maps;
+import com.google.common.collect.Sets;
+import org.apache.commons.io.IOUtils;
+import org.apache.commons.lang.StringUtils;
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.apache.oozie.client.OozieClient;
+import org.apache.oozie.client.OozieClientException;
+import org.apache.oozie.client.WorkflowAction;
+import org.apache.oozie.client.WorkflowJob;
+import org.apache.oozie.client.WorkflowJob.Status;
+
+import eu.dnetlib.data.hadoop.action.JobCompletion;
+import eu.dnetlib.data.hadoop.action.JobMonitor;
+import eu.dnetlib.data.hadoop.rmi.HadoopServiceException;
+
+import static java.lang.String.format;
+
+public class OozieJobMonitor extends JobMonitor {
+
+	private static final Log log = LogFactory.getLog(JobMonitor.class); // NOPMD by marko on 11/24/08 5:02 PM
+
+	private final OozieClient oozieClient;
+
+	private final String jobId;
+
+	public static final String ACTION_TYPE_SUBWORKFLOW = "sub-workflow";
+
+	private Set<String> workflowActions = Sets.newHashSet();
+
+	@Deprecated
+	public OozieJobMonitor(final OozieClient oozieClient, String jobId, final JobCompletion callback) {
+		super(callback);
+		this.oozieClient = oozieClient;
+		this.jobId = jobId;
+	}
+
+	public OozieJobMonitor(final OozieClient oozieClient, String jobId, final JobCompletion callback, final Set<String> workflowActions) {
+		super(callback);
+		this.oozieClient = oozieClient;
+		this.jobId = jobId;
+		this.workflowActions = workflowActions;
+	}
+
+	@Override
+	public void run() {
+		try {
+			log.info("waiting for oozie job completion: " + getHadoopId());
+
+			Status status = Status.PREP;
+			while (status.equals(Status.PREP) || status.equals(Status.RUNNING)) {
+				Thread.sleep(monitorSleepTimeSecs * 1000);
+
+				try {
+					final Status currentStatus = doGetStatus();
+
+					if (!currentStatus.equals(status)) {
+						status = currentStatus;
+						lastActivity = new Date();
+					}
+				} catch (Throwable e) {
+					log.warn(format("error polling status for job %s", jobId), e);
+				}
+			}
+
+			log.debug(format("job %s finihsed with status: %s", jobId, status));
+			if (Status.SUCCEEDED.equals(status)) {
+				// TODO set some content to return to the blackboard msg.
+
+				log.info(format("looking for oozie job(%s) output values: %s", getHadoopId(), workflowActions));
+				final Properties report = getReport(getOozieClient(), getHadoopId(), workflowActions);
+				if (report != null) {
+					final Map<String, String> map = Maps.newHashMap();
+					report.forEach((k, v) -> map.put(k.toString(), v.toString()));
+					log.info("found oozie job report, size: " + map.size());
+					getCallback().done(map);
+				} else {
+					log.warn("cannot find oozie job report!");
+					getCallback().done(new HashMap<>());
+				}
+            } else {
+				// TODO retrieve some failure information from the oozie client.
+				String msg = format("hadoop job: %s failed with status: %s, oozie log:\n %s\n", getHadoopId(), getStatus(), getOozieClient().getJobLog(getHadoopId()));
+				getCallback().failed(msg, new HadoopServiceException(msg));
+            }
+		} catch (Throwable e) {
+			getCallback().failed(getHadoopId(), e);
+		}
+	}
+
+	/**
+	 * Provides report entries when found for given oozie job identifier. Returns null when report not found.
+	 */
+	private static Properties getReport(final OozieClient oozieClient, final String oozieJobId, final Set<String> workflowActions) throws OozieClientException, IOException {
+		WorkflowJob oozieJob = oozieClient.getJobInfo(oozieJobId);
+		for (WorkflowAction currentAction : oozieJob.getActions()) {
+			log.info(String.format("looking for workflow actions to report, current: '%s'", currentAction.getName()));
+			if (workflowActions.contains(currentAction.getName())) {
+				log.info(String.format("found workflow action %s", currentAction.getName()));
+				if (ACTION_TYPE_SUBWORKFLOW.equals(currentAction.getType())) {
+					log.info(String.format("looking for sub-workflow actions external id: %s", currentAction.getExternalId()));
+					Properties subworkflowProperties = getReport(oozieClient, currentAction.getExternalId(), workflowActions);
+					if (subworkflowProperties != null) {
+						return subworkflowProperties;
+					}
+				} else if (StringUtils.isNotBlank(currentAction.getData())) {
+					Properties properties = new Properties();
+					properties.load(IOUtils.toInputStream(currentAction.getData()));
+					log.info(String.format("found workflow action(%s) properties size %s", currentAction.getName(), properties.values().size()));
+					return properties;
+				}
+			} else {
+				log.info(String.format("cannot find workflow action(%s) properties", currentAction.getName()));
+			}
+		}
+		return null;
+	}
+
+	@Override
+	public String getHadoopId() {
+		return jobId;
+	}
+
+	public OozieClient getOozieClient() {
+		return oozieClient;
+	}
+
+	@Override
+	public String getStatus() {
+		try {
+			return doGetStatus().toString();
+		} catch (OozieClientException e) {
+			log.error("error accessing job status", e);
+			return "UNKNOWN";
+		}
+	}
+
+	private Status doGetStatus() throws OozieClientException {
+		return getOozieClient().getJobInfo(getHadoopId()).getStatus();
+	}
+
+	@Override
+	public Date getLastActivity() {
+		return lastActivity;
+	}
+
+	@Override
+	public Date getStartTime() throws HadoopServiceException {
+		try {
+			return getOozieClient().getJobInfo(getHadoopId()).getStartTime();
+		} catch (OozieClientException e) {
+			throw new HadoopServiceException("unable to read job start time", e);
+		}
+	}
+
+	@Override
+	public String getTrackerUrl() {
+		return getOozieClient().getOozieUrl();
+	}
+
+	@Override
+	public void kill() {
+		try {
+			getOozieClient().kill(getHadoopId());
+		} catch (OozieClientException e) {
+			log.error("unable to kill job: " + getHadoopId(), e);
+		}
+	}
+
+}
Index: modules/dnet-hadoop-service/tags/dnet-hadoop-service-2.7.8/src/main/java/eu/dnetlib/data/hadoop/oozie/OozieClientFactory.java
===================================================================
--- modules/dnet-hadoop-service/tags/dnet-hadoop-service-2.7.8/src/main/java/eu/dnetlib/data/hadoop/oozie/OozieClientFactory.java	(nonexistent)
+++ modules/dnet-hadoop-service/tags/dnet-hadoop-service-2.7.8/src/main/java/eu/dnetlib/data/hadoop/oozie/OozieClientFactory.java	(revision 59871)
@@ -0,0 +1,35 @@
+package eu.dnetlib.data.hadoop.oozie;
+
+import eu.dnetlib.data.hadoop.AbstractHadoopClient;
+import eu.dnetlib.data.hadoop.config.ClusterName;
+import eu.dnetlib.data.hadoop.rmi.HadoopServiceException;
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.apache.oozie.client.OozieClient;
+
+/**
+ * Factory bean for Oozie client instances.
+ * 
+ * oozie server may not be available in every hadoop deployment, so every oozie specific operation should be defined as optional.
+ * 
+ * @author claudio
+ * 
+ */
+public class OozieClientFactory extends AbstractHadoopClient {
+
+	private static final String ENV_ATTRIBUTE_OOZIE_SERVICE_LOC = "oozie.service.loc";
+
+	private static final Log log = LogFactory.getLog(OozieClientFactory.class); // NOPMD by marko on 11/24/08 5:02 PM
+
+	public OozieClient newInstance(ClusterName clusterName) throws HadoopServiceException {
+		final String oozieServiceLocation = configurationEnumerator.get(clusterName).get(ENV_ATTRIBUTE_OOZIE_SERVICE_LOC);
+		log.info("init oozie client, cluster: " + clusterName.toString() + ", oozie server: " + oozieServiceLocation);
+		setHadoopUser();
+		try {
+			return new OozieClient(oozieServiceLocation);
+		} catch (Throwable e) {
+			throw new HadoopServiceException("unable to initialize oozie client for cluster: " + clusterName.toString(), e);
+		}
+	}
+
+}
Index: modules/dnet-hadoop-service/tags/dnet-hadoop-service-2.7.8/src/main/java/eu/dnetlib/data/hadoop/utils/ScanFactory.java
===================================================================
--- modules/dnet-hadoop-service/tags/dnet-hadoop-service-2.7.8/src/main/java/eu/dnetlib/data/hadoop/utils/ScanFactory.java	(nonexistent)
+++ modules/dnet-hadoop-service/tags/dnet-hadoop-service-2.7.8/src/main/java/eu/dnetlib/data/hadoop/utils/ScanFactory.java	(revision 59871)
@@ -0,0 +1,89 @@
+package eu.dnetlib.data.hadoop.utils;
+
+import java.io.ByteArrayOutputStream;
+import java.io.DataOutputStream;
+import java.io.IOException;
+import java.util.Map;
+
+import org.apache.commons.lang.StringUtils;
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.apache.hadoop.hbase.client.Scan;
+import org.apache.hadoop.hbase.filter.PrefixFilter;
+import org.apache.hadoop.hbase.util.Base64;
+import org.dom4j.Document;
+import org.dom4j.Node;
+
+public class ScanFactory {
+
+	private static final Log log = LogFactory.getLog(ScanFactory.class); // NOPMD by marko on 11/24/08 5:02 PM
+
+	public static String getScan(final ScanProperties scanProperties) throws IOException {
+		Scan scan = new Scan();
+
+		scan.setCaching(scanProperties.getCaching());
+		scan.setCacheBlocks(false); // don't set to true for MR jobs
+
+		scan.setFilter(scanProperties.getFilterList());
+		for (String family : scanProperties.getFamilies()) {
+			scan.addFamily(family.getBytes());
+		}
+
+		log.debug("serializing scan");
+		return convertScanToString(scan);
+	}
+
+	public static ScanProperties parseScanProperties(final Document doc, final Map<String, String> bbParams) {
+		log.debug("setting job scanner");
+
+		ScanProperties scanProperties = new ScanProperties(doc.valueOf("//FILTERS/@operator"));
+
+		String caching = doc.valueOf("//SCAN/@caching");
+		if (!StringUtils.isBlank(caching)) {
+			log.info("overriding default scan caching with: " + caching);
+			scanProperties.setCaching(Integer.valueOf(caching));
+		}
+
+		for (Object o : doc.selectNodes("//SCAN/FAMILIES/FAMILY")) {
+			Node node = (Node) o;
+			String value = node.valueOf("./@value");
+			if ((value == null) || value.isEmpty()) {
+				value = bbParams.get(node.valueOf("./@param"));
+			}
+			log.debug("scanner family value: " + value);
+			scanProperties.getFamilies().add(value);
+		}
+		for (Object o : doc.selectNodes("//SCAN/FILTERS/FILTER")) {
+			Node node = (Node) o;
+			String filterType = node.valueOf("./@type");
+
+			String value = node.valueOf("./@value");
+			if ((value == null) || value.isEmpty()) {
+				value = bbParams.get(node.valueOf("./@param"));
+			}
+
+			if (filterType.equals("prefix")) {
+				log.debug("scanner prefix filter, value: " + value);
+				scanProperties.getFilterList().addFilter(new PrefixFilter(value.getBytes()));
+			} // TODO add more filterType cases here
+		}
+		return scanProperties;
+	}
+
+	/**
+	 * Writes the given scan into a Base64 encoded string.
+	 *
+	 * @param scan
+	 *            The scan to write out.
+	 * @return The scan saved in a Base64 encoded string.
+	 * @throws IOException
+	 *             When writing the scan fails.
+	 */
+	private static String convertScanToString(final Scan scan) throws IOException {
+		ByteArrayOutputStream out = new ByteArrayOutputStream();
+		DataOutputStream dos = new DataOutputStream(out);
+		scan.write(dos);
+		return Base64.encodeBytes(out.toByteArray());
+	}
+
+}
Index: modules/dnet-hadoop-service/tags/dnet-hadoop-service-2.7.8/src/main/java/eu/dnetlib/data/hadoop/HadoopClientMap.java
===================================================================
--- modules/dnet-hadoop-service/tags/dnet-hadoop-service-2.7.8/src/main/java/eu/dnetlib/data/hadoop/HadoopClientMap.java	(nonexistent)
+++ modules/dnet-hadoop-service/tags/dnet-hadoop-service-2.7.8/src/main/java/eu/dnetlib/data/hadoop/HadoopClientMap.java	(revision 59871)
@@ -0,0 +1,117 @@
+package eu.dnetlib.data.hadoop;
+
+import java.io.IOException;
+import java.util.Map;
+
+import com.google.common.collect.Maps;
+import com.google.gson.Gson;
+import eu.dnetlib.data.hadoop.config.ClusterName;
+import eu.dnetlib.data.hadoop.hbase.HBaseAdminFactory;
+import eu.dnetlib.data.hadoop.mapred.JobClientFactory;
+import eu.dnetlib.data.hadoop.oozie.OozieClientFactory;
+import eu.dnetlib.data.hadoop.rmi.HadoopServiceException;
+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.Required;
+
+
+public class HadoopClientMap {
+
+	private static final Log log = LogFactory.getLog(HadoopClientMap.class); // NOPMD by marko on 11/24/08 5:02 PM
+
+	private JobClientFactory jobClientFactory;
+
+	private OozieClientFactory oozieClientFactory;
+
+	private HBaseAdminFactory hbaseAdminFactory;
+
+	private Map<String, Map<String, String>> enabledClients = Maps.newHashMap();
+
+	public boolean isMapreduceAvailable(final ClusterName name) {
+		return isClientAvailable(name, "mapred");
+	}
+
+	public boolean isOozieAvailable(final ClusterName name) {
+		return isClientAvailable(name, "oozie");
+	}
+
+	public boolean isHbaseAvailable(final ClusterName name) {
+		return isClientAvailable(name, "hbase");
+	}
+
+	private boolean isClientAvailable(final ClusterName name, final String clientName) {
+		final String clusterName = name.toString();
+		return enabledClients.containsKey(clusterName) && "true".equals(enabledClients.get(clusterName).get(clientName));
+	}
+
+	public JobClient getJtClient(final ClusterName clusterName, final String username) throws HadoopServiceException, IOException {
+		if (!isMapreduceAvailable(clusterName)) {
+			throw new HadoopServiceException("jobtracker is not available for cluster " + clusterName.toString());
+		}
+		return getJobClientFactory().newInstance(clusterName, username);
+	}
+
+	public JobClient getJtClient(final ClusterName clusterName) throws HadoopServiceException, IOException {
+		if (!isMapreduceAvailable(clusterName)) {
+			throw new HadoopServiceException("jobtracker is not available for cluster " + clusterName.toString());
+		}
+		return getJobClientFactory().newInstance(clusterName);
+	}
+
+	public OozieClient getOozieClient(final ClusterName name) throws HadoopServiceException {
+		if (!isOozieAvailable(name)) {
+			throw new HadoopServiceException("oozie is not available for cluster " + name.toString());
+		}
+		return getOozieClientFactory().newInstance(name);
+	}
+
+	public HBaseAdmin getHbaseAdmin(final ClusterName name) throws HadoopServiceException {
+		if (!isHbaseAvailable(name)) {
+			throw new HadoopServiceException("hbase is not available for cluster " + name.toString());
+		}
+		return getHbaseAdminFactory().newInstance(name);
+	}
+
+	// //////////
+
+	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;
+	}
+
+	@Required
+	public void setJobClientFactory(final JobClientFactory jobClientFactory) {
+		this.jobClientFactory = jobClientFactory;
+	}
+
+	public OozieClientFactory getOozieClientFactory() {
+		return oozieClientFactory;
+	}
+
+	@Required
+	public void setOozieClientFactory(final OozieClientFactory oozieClientFactory) {
+		this.oozieClientFactory = oozieClientFactory;
+	}
+
+	public HBaseAdminFactory getHbaseAdminFactory() {
+		return hbaseAdminFactory;
+	}
+
+	@Required
+	public void setHbaseAdminFactory(final HBaseAdminFactory hbaseAdminFactory) {
+		this.hbaseAdminFactory = hbaseAdminFactory;
+	}
+
+}
Index: modules/dnet-hadoop-service/tags/dnet-hadoop-service-2.7.8/src/main/java/eu/dnetlib/data/hadoop/JobRegistry.java
===================================================================
--- modules/dnet-hadoop-service/tags/dnet-hadoop-service-2.7.8/src/main/java/eu/dnetlib/data/hadoop/JobRegistry.java	(nonexistent)
+++ modules/dnet-hadoop-service/tags/dnet-hadoop-service-2.7.8/src/main/java/eu/dnetlib/data/hadoop/JobRegistry.java	(revision 59871)
@@ -0,0 +1,99 @@
+package eu.dnetlib.data.hadoop;
+
+import java.util.Date;
+import java.util.List;
+import java.util.Map.Entry;
+import java.util.Objects;
+import java.util.stream.Collectors;
+
+import com.google.common.collect.BiMap;
+import com.google.common.collect.HashBiMap;
+import eu.dnetlib.data.hadoop.HadoopJob.Status;
+import eu.dnetlib.data.hadoop.config.ClusterName;
+import eu.dnetlib.data.hadoop.rmi.HadoopJobDescriptor;
+import eu.dnetlib.data.hadoop.rmi.HadoopServiceException;
+import eu.dnetlib.data.hadoop.utils.HadoopUtils;
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.springframework.beans.factory.annotation.Required;
+
+public class JobRegistry {
+
+	private static final Log log = LogFactory.getLog(JobRegistry.class); // NOPMD by marko on 11/24/08 5:02 PM
+
+	private int maxJobs;
+
+	private final BiMap<String, HadoopJob> jobs = HashBiMap.create();
+
+	public String registerJob(HadoopJob hadoopJob) throws HadoopServiceException {
+
+		if (jobs.containsValue(hadoopJob)) { return jobs.inverse().get(hadoopJob); }
+
+		if (jobs.size() > getMaxJobs()) {
+			removeOldestProcess();
+		}
+
+		jobs.put(hadoopJob.getId(), hadoopJob);
+		log.info("Registered hadoop job " + hadoopJob.getId());
+		hadoopJob.startMonitor();
+
+		return hadoopJob.getId();
+	}
+
+	public Status getJobStatus(String id) {
+		return findJob(id).getStatus();
+	}
+
+	public HadoopJob findJob(String id) {
+		return jobs.get(id);
+	}
+
+	public void unregisterJob(String id) throws HadoopServiceException {
+
+		if (!jobs.containsKey(id)) { throw new HadoopServiceException("unable to unregister job, could not find jobId in registry: " + id); }
+
+		log.info("unregistering job: " + id);
+		jobs.get(id).getJobMonitor().kill();
+		jobs.remove(id);
+	}
+
+	private void removeOldestProcess() throws HadoopServiceException {
+		Date oldDate = new Date();
+		String oldId = null;
+
+		for (Entry<String, HadoopJob> e : jobs.entrySet()) {
+			final HadoopJob hadoopJob = e.getValue();
+
+			if (hadoopJob.isComplete()) {
+				final Date date = hadoopJob.getLastActivity();
+				if (date.before(oldDate)) {
+					oldDate = date;
+					oldId = e.getKey();
+				}
+			}
+		}
+
+		if (oldId != null) {
+			unregisterJob(oldId);
+		}
+
+	}
+
+	public List<HadoopJobDescriptor> listJobs(ClusterName clusterName) {
+		return jobs.values().stream()
+				.filter(j -> clusterName.equals(j.getClusterName()))
+				.map(HadoopUtils.asDescriptor())
+				.filter(Objects::nonNull)
+				.collect(Collectors.toList());
+	}
+
+	@Required
+	public void setMaxJobs(final int maxJobs) {
+		this.maxJobs = maxJobs;
+	}
+
+	public int getMaxJobs() {
+		return maxJobs;
+	}
+
+}
Index: modules/dnet-hadoop-service/tags/dnet-hadoop-service-2.7.8/src/main/java/eu/dnetlib/data/hadoop/ISClient.java
===================================================================
--- modules/dnet-hadoop-service/tags/dnet-hadoop-service-2.7.8/src/main/java/eu/dnetlib/data/hadoop/ISClient.java	(nonexistent)
+++ modules/dnet-hadoop-service/tags/dnet-hadoop-service-2.7.8/src/main/java/eu/dnetlib/data/hadoop/ISClient.java	(revision 59871)
@@ -0,0 +1,73 @@
+package eu.dnetlib.data.hadoop;
+
+import java.util.List;
+
+import javax.annotation.Resource;
+
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+
+import com.google.common.collect.Iterables;
+
+import eu.dnetlib.enabling.is.lookup.rmi.ISLookUpException;
+import eu.dnetlib.enabling.is.lookup.rmi.ISLookUpService;
+import eu.dnetlib.enabling.is.registry.rmi.ISRegistryException;
+import eu.dnetlib.enabling.is.registry.rmi.ISRegistryService;
+import eu.dnetlib.enabling.locators.UniqueServiceLocator;
+import eu.dnetlib.miscutils.datetime.DateUtils;
+
+public class ISClient {
+
+	private static final Log log = LogFactory.getLog(ISClient.class); // NOPMD by marko on 11/24/08 5:02 PM
+
+	@Resource
+	private UniqueServiceLocator serviceLocator;
+
+	public String getJobProfile(final String jobName) throws ISLookUpException {
+		return serviceLocator.getService(ISLookUpService.class).getResourceProfileByQuery(
+				"/RESOURCE_PROFILE[.//RESOURCE_TYPE/@value = 'HadoopJobConfigurationDSResourceType' and .//HADOOP_JOB/@name='" + jobName + "']");
+	}
+
+	public String queryForServiceProperty(final String key) throws ISLookUpException {
+		return getServiceConfigValue(
+				String.format(
+						"distinct-values("
+								+ "for $x in /RESOURCE_PROFILE[.//RESOURCE_TYPE/@value='HadoopServiceResourceType'] "
+								+ "return $x//SERVICE_PROPERTIES/PROPERTY[./@ key='%s']/@value/string())",
+						key));
+	}
+
+	public void updateCountElement(final String jobName, final String element, final String delta) {
+		final String xquery =
+				"let $x := //RESOURCE_PROFILE[" + ".//RESOURCE_TYPE/@value='HadoopJobConfigurationDSResourceType' and .//HADOOP_JOB/@name='" + jobName
+						+ "'], $tot := $x//STATUS/" + element + "/@value/number() " + delta + " return update replace $x//STATUS/" + element + " with <"
+						+ element + " value='{$tot}' />";
+
+		executeXUpdate(xquery);
+	}
+
+	public void updateDate(final String jobName) {
+		log.info("increment last submission date for job: " + jobName);
+		executeXUpdate("for $x in collection('')/RESOURCE_PROFILE["
+				+ ".//RESOURCE_TYPE/@value='HadoopJobConfigurationDSResourceType' and .//HADOOP_JOB/@name='"
+				+ jobName + "'] " + " return update value $x//LAST_SUBMISSION_DATE/@value with '" + DateUtils.now_ISO8601() + "' ");
+	}
+
+	private String getServiceConfigValue(final String xquery) throws ISLookUpException {
+		log.debug("quering for service property: " + xquery);
+		final List<String> urls = serviceLocator.getService(ISLookUpService.class).quickSearchProfile(xquery);
+		if ((urls == null) || (urls.size() != 1))
+			throw new IllegalStateException("unable to find unique service property, xquery: " + xquery);
+		return Iterables.getOnlyElement(urls);
+	}
+
+	private boolean executeXUpdate(final String xupdate) {
+		try {
+			log.debug("running xupdate: " + xupdate);
+			return serviceLocator.getService(ISRegistryService.class).executeXUpdate(xupdate);
+		} catch (final ISRegistryException e) {
+			log.error("unable to run xupdate: " + xupdate, e);
+			return false;
+		}
+	}
+}
Index: modules/dnet-hadoop-service/tags/dnet-hadoop-service-2.7.8/src/test/java/eu/dnetlib/data/hadoop/hbase/HBaseTest.java
===================================================================
--- modules/dnet-hadoop-service/tags/dnet-hadoop-service-2.7.8/src/test/java/eu/dnetlib/data/hadoop/hbase/HBaseTest.java	(nonexistent)
+++ modules/dnet-hadoop-service/tags/dnet-hadoop-service-2.7.8/src/test/java/eu/dnetlib/data/hadoop/hbase/HBaseTest.java	(revision 59871)
@@ -0,0 +1,106 @@
+package eu.dnetlib.data.hadoop.hbase;
+
+import java.io.IOException;
+import java.util.Map.Entry;
+import java.util.NavigableMap;
+import java.util.Set;
+
+import com.google.common.collect.Sets;
+import eu.dnetlib.data.hadoop.HadoopServiceCore;
+import eu.dnetlib.data.hadoop.config.ClusterName;
+import eu.dnetlib.data.hadoop.config.ConfigurationEnumerator;
+import eu.dnetlib.data.hadoop.rmi.HadoopServiceException;
+import eu.dnetlib.miscutils.datetime.DateUtils;
+import org.apache.hadoop.hbase.client.Get;
+import org.apache.hadoop.hbase.client.HTable;
+import org.apache.hadoop.hbase.client.Put;
+import org.apache.hadoop.hbase.client.Result;
+import org.apache.hadoop.hbase.util.Bytes;
+import org.junit.After;
+import org.junit.Before;
+import org.junit.Ignore;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.test.context.ActiveProfiles;
+import org.springframework.test.context.ContextConfiguration;
+import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
+
+import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.assertTrue;
+
+@ActiveProfiles("test")
+@RunWith(SpringJUnit4ClassRunner.class)
+@ContextConfiguration(classes = HBaseTestContextConfiguration.class)
+public class HBaseTest {
+
+	protected static final String TEST_TABLE = "dnet_test_table";
+
+	protected static final int NUM_VERSIONS = 10;
+
+	@Autowired
+	private HadoopServiceCore hadoopServiceCore;
+
+	@Autowired
+	private ConfigurationEnumerator configurationEnumerator;
+
+	@Before
+	public void setUp() throws HadoopServiceException, IOException, InterruptedException {
+		assertNotNull(hadoopServiceCore);
+
+		ensureDropTable();
+	}
+
+	@After
+	public void tearDown() throws HadoopServiceException, IOException {
+		ensureDropTable();
+	}
+
+	@Test
+	@Ignore
+	// TODO allow testing on a dev cluster instance
+	public void testReadWrite() throws HadoopServiceException, IOException, InterruptedException {
+
+		hadoopServiceCore.createTable(ClusterName.DM, TEST_TABLE, testSchema());
+		assertTrue(hadoopServiceCore.existTable(ClusterName.DM, TEST_TABLE));
+
+		final HTable htable = new HTable(configurationEnumerator.get(ClusterName.DM), TEST_TABLE);
+
+		final Put put = new Put(Bytes.toBytes("1"));
+
+		for (int i = 0; i < NUM_VERSIONS; i++) {
+			put.add(Bytes.toBytes("result"), Bytes.toBytes("body"), Bytes.toBytes(i + ""));
+			htable.put(put);
+			Thread.sleep(1000);
+		}
+		final Get get = new Get(Bytes.toBytes("1"));
+		get.setMaxVersions(HBaseTestContextConfiguration.MAX_VERSIONS);
+
+		final Result r = htable.get(get);
+
+		// Map<family,Map<qualifier,Map<timestamp,value>>>
+		final NavigableMap<Long, byte[]> versions = r.getMap().get(Bytes.toBytes("result")).get(Bytes.toBytes("body"));
+
+		for (final Entry<Long, byte[]> e : versions.entrySet()) {
+			System.out.println("t: " + DateUtils.calculate_ISO8601(e.getKey()) + ", v: " + Bytes.toString(e.getValue()));
+		}
+
+		htable.close();
+
+	}
+
+	protected void ensureDropTable() throws HadoopServiceException, IOException {
+		if (hadoopServiceCore.existTable(ClusterName.DM, TEST_TABLE)) {
+			hadoopServiceCore.dropTable(ClusterName.DM, TEST_TABLE);
+		}
+	}
+
+	protected Set<String> testSchema() {
+		final Set<String> schema = Sets.newHashSet();
+
+		schema.add("result");
+
+		return schema;
+	}
+
+}
Index: modules/dnet-hadoop-service/tags/dnet-hadoop-service-2.7.8/src/main/java/eu/dnetlib/data/hadoop/hbase/HBaseAdminFactory.java
===================================================================
--- modules/dnet-hadoop-service/tags/dnet-hadoop-service-2.7.8/src/main/java/eu/dnetlib/data/hadoop/hbase/HBaseAdminFactory.java	(nonexistent)
+++ modules/dnet-hadoop-service/tags/dnet-hadoop-service-2.7.8/src/main/java/eu/dnetlib/data/hadoop/hbase/HBaseAdminFactory.java	(revision 59871)
@@ -0,0 +1,24 @@
+package eu.dnetlib.data.hadoop.hbase;
+
+import eu.dnetlib.data.hadoop.AbstractHadoopClient;
+import eu.dnetlib.data.hadoop.config.ClusterName;
+import eu.dnetlib.data.hadoop.rmi.HadoopServiceException;
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.apache.hadoop.hbase.client.HBaseAdmin;
+
+public class HBaseAdminFactory extends AbstractHadoopClient {
+
+	private static final Log log = LogFactory.getLog(HBaseAdminFactory.class); // NOPMD by marko on 11/24/08 5:02 PM
+
+	public HBaseAdmin newInstance(final ClusterName clusterName) throws HadoopServiceException {
+		try {
+			log.info("init hbaseAdmin, cluster: " + clusterName.toString());
+			setHadoopUser();
+			return new HBaseAdmin(configurationEnumerator.get(clusterName));
+		} catch (final Throwable e) {
+			throw new HadoopServiceException("unable to initialize hbase client for cluster: " + clusterName.toString(), e);
+		}
+	}
+
+}
Index: modules/dnet-hadoop-service/tags/dnet-hadoop-service-2.7.8/src/test/resources/log4j.properties
===================================================================
--- modules/dnet-hadoop-service/tags/dnet-hadoop-service-2.7.8/src/test/resources/log4j.properties	(nonexistent)
+++ modules/dnet-hadoop-service/tags/dnet-hadoop-service-2.7.8/src/test/resources/log4j.properties	(revision 59871)
@@ -0,0 +1,53 @@
+### Root Level ###
+log4j.rootLogger=WARN, LOGFILE, CONSOLE
+
+### Configuration for the LOGFILE appender ###
+log4j.appender.LOGFILE=org.apache.log4j.RollingFileAppender
+log4j.appender.LOGFILE.MaxFileSize=25MB
+log4j.appender.LOGFILE.MaxBackupIndex=10
+log4j.appender.LOGFILE.File=logs/dnet.log
+log4j.appender.LOGFILE.Append=true
+log4j.appender.LOGFILE.layout=org.apache.log4j.PatternLayout
+log4j.appender.LOGFILE.layout.ConversionPattern=[%-5p] %d %c - %m%n
+
+### Configuration for the CONSOLE appender ###
+log4j.appender.CONSOLE=org.apache.log4j.ConsoleAppender
+log4j.appender.CONSOLE.layout=org.apache.log4j.PatternLayout
+log4j.appender.CONSOLE.layout.ConversionPattern=[%-5p] %d %c - %m%n
+
+org.apache.cxf.Logger=org.apache.cxf.common.logging.Log4jLogger
+
+### Application Level ###
+log4j.logger.eu.dnetlib=INFO
+log4j.logger.eu.dnetlib.enabling.is.sn=INFO
+log4j.logger.org.apache.cxf.interceptor=FATAL
+log4j.logger.org.apache.cxf.ws.addressing.ContextUtils=FATAL
+log4j.logger.eu.dnetlib.enabling.tools.AbstractBaseService=INFO
+log4j.logger.eu.dnetlib.enabling.inspector=DEBUG
+log4j.logger.eu.dnetlib.xml.database.LoggingTrigger=WARN
+log4j.logger.eu.dnetlib.enabling.tools.registration.ServiceRegistrator=INFO
+log4j.logger.eu.dnetlib.enabling.inspector=FATAL
+log4j.logger.eu.dnetlib.enabling.inspector.SubscriptionController=DEBUG
+log4j.logger.eu.dnetlib.springutils.stringtemplate.StringTemplateViewResolver=FATAL             
+log4j.logger.eu.dnetlib.enabling.is.sn.SynchronousNotificationSenderImpl=WARN
+log4j.logger.eu.dnetlib.enabling.tools.LocalServiceResolverImpl=WARN
+log4j.logger.eu.dnetlib.enabling.is.sn.NotificationInvokerImpl=WARN
+log4j.logger.eu.dnetlib.data.collective=INFO
+log4j.logger.eu.dnetlib.data.hadoop.utils.ScanFactory=DEBUG
+log4j.logger.org.apache.xerces.parsers.SAXParser=OFF
+log4j.logger.eu.dnetlib.conf.PropertyFetcher=WARN
+#log4j.logger.eu.dnetlib.data.transform.XsltRowTransformerFactory=DEBUG
+
+log4j.logger.eu.dnetlib.enabling.is.sn.ISSNServiceImpl=OFF
+log4j.logger.eu.dnetlib.enabling.datasources.DatasourceManagerClients=FATAL
+log4j.logger.eu.dnetlib.data.mdstore.modular.mongodb.utils.MetadataCheckJob=DEBUG
+log4j.logger.eu.dnetlib.enabling.is.sn.ISSNServiceCore=WARN
+log4j.logger.eu.dnetlib.xml.database.exist.ExistDatabase=WARN
+log4j.logger.eu.dnetlib.enabling.is.store.AbstractContentInitializer=FATAL
+
+log4j.logger.org.apache.hadoop.hbase.mapreduce.TableInputFormatBase=FATAL
+
+### Spring ###
+log4j.logger.org.springframework=ERROR
+
+
Index: modules/dnet-hadoop-service/tags/dnet-hadoop-service-2.7.8/src/main/java/eu/dnetlib/data/hadoop/hbase/HbaseTableFeeder.java
===================================================================
--- modules/dnet-hadoop-service/tags/dnet-hadoop-service-2.7.8/src/main/java/eu/dnetlib/data/hadoop/hbase/HbaseTableFeeder.java	(nonexistent)
+++ modules/dnet-hadoop-service/tags/dnet-hadoop-service-2.7.8/src/main/java/eu/dnetlib/data/hadoop/hbase/HbaseTableFeeder.java	(revision 59871)
@@ -0,0 +1,181 @@
+package eu.dnetlib.data.hadoop.hbase;
+
+import java.io.IOException;
+import java.util.Arrays;
+import java.util.List;
+
+import com.google.common.base.Predicates;
+import com.google.common.collect.Iterables;
+import com.google.common.collect.Lists;
+import eu.dnetlib.data.hadoop.config.ClusterName;
+import eu.dnetlib.data.hadoop.config.ConfigurationEnumerator;
+import eu.dnetlib.data.transform.Row;
+import eu.dnetlib.data.transform.XsltRowTransformerFactory;
+import eu.dnetlib.enabling.resultset.client.ResultSetClientFactory;
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.hbase.client.HTable;
+import org.apache.hadoop.hbase.client.Mutation;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.beans.factory.annotation.Required;
+
+/**
+ * The Class HbaseTableFeeder provides abstraction to ship batch operation on an HBase table
+ */
+public abstract class HbaseTableFeeder {
+
+	/**
+	 * The Constant log.
+	 */
+	private static final Log log = LogFactory.getLog(HbaseTableFeeder.class); // NOPMD by marko on 11/24/08 5:02 PM
+	/**
+	 * The configuration enumerator.
+	 */
+	@Autowired
+	protected ConfigurationEnumerator configurationEnumerator;
+	/**
+	 * The batch size.
+	 */
+	private int batchSize = 100;
+	/**
+	 * The result set client factory.
+	 */
+	private ResultSetClientFactory resultSetClientFactory;
+
+	/**
+	 * Adds the operation.
+	 *
+	 * @param buffer the buffer
+	 * @param row    the row
+	 */
+	protected abstract void addOperation(final List<Mutation> buffer, final Row row);
+
+	/**
+	 * Feed.
+	 *
+	 * @param epr         the epr
+	 * @param xsl         the xsl
+	 * @param clusterName the cluster name
+	 * @param tableName   the table name
+	 * @param simulation  the simulation
+	 * @return the int
+	 * @throws IOException          Signals that an I/O exception has occurred.
+	 * @throws InterruptedException the interrupted exception
+	 */
+	public int feed(final String epr, final String xsl, final ClusterName clusterName, final String tableName, final boolean simulation)
+			throws IOException, InterruptedException {
+		return doWrite(asRows(epr, xsl), getConf(clusterName), tableName, simulation);
+	}
+
+	/**
+	 * Do writeOnHBase.
+	 *
+	 * @param rows          the rows
+	 * @param configuration the configuration
+	 * @param tableName     the table name
+	 * @param simulation    the simulation
+	 * @return the int
+	 * @throws IOException          Signals that an I/O exception has occurred.
+	 * @throws InterruptedException the interrupted exception
+	 */
+	private int doWrite(final Iterable<Row> rows, final Configuration configuration, final String tableName, final boolean simulation)
+			throws IOException, InterruptedException {
+		final List<Mutation> buffer = Lists.newArrayList();
+
+		int count = 0;
+		if (simulation) {
+			log.info("running in simulation mode ...");
+			log.info(String.format("... simulated import of %d records", Iterables.size(rows)));
+		} else {
+
+			final HTable htable = new HTable(configuration, tableName);
+			try {
+				int i = 0;
+				for (final Row row : rows) {
+					addOperation(buffer, row);
+					if ((++i % getBatchSize()) == 0) {
+						flush(tableName, buffer, htable);
+						count += buffer.size();
+						buffer.clear();
+					}
+				}
+			} finally {
+				if (!buffer.isEmpty()) {
+					flush(tableName, buffer, htable);
+					count += buffer.size();
+				}
+				htable.flushCommits();
+				htable.close();
+			}
+		}
+		return count;
+	}
+
+	private void flush(final String tableName, final List<Mutation> buffer, final HTable htable) throws IOException, InterruptedException {
+		if (!checkOp(htable.batch(buffer), tableName)) throw new IOException("unable to flush operation on HBase table: " + tableName);
+	}
+
+	private boolean checkOp(final Object[] res, final String tableName) throws IOException {
+		return Iterables.all(Arrays.asList(res), Predicates.notNull());
+	}
+
+	/**
+	 * As rows.
+	 *
+	 * @param epr the epr
+	 * @param xsl the xsl
+	 * @return the iterable
+	 */
+	protected Iterable<Row> asRows(final String epr, final String xsl) {
+		return Iterables.concat(Iterables.transform(getResultSetClientFactory().getClient(epr), XsltRowTransformerFactory.newInstance(xsl)));
+	}
+
+	/**
+	 * Gets the conf.
+	 *
+	 * @param clusterName the cluster name
+	 * @return the conf
+	 */
+	protected Configuration getConf(final ClusterName clusterName) {
+		return configurationEnumerator.get(clusterName);
+	}
+
+	/**
+	 * Gets the batch size.
+	 *
+	 * @return the batch size
+	 */
+	public int getBatchSize() {
+		return batchSize;
+	}
+
+	/**
+	 * Sets the batch size.
+	 *
+	 * @param batchSize the new batch size
+	 */
+	public void setBatchSize(final int batchSize) {
+		this.batchSize = batchSize;
+	}
+
+	/**
+	 * Gets the result set client factory.
+	 *
+	 * @return the result set client factory
+	 */
+	public ResultSetClientFactory getResultSetClientFactory() {
+		return resultSetClientFactory;
+	}
+
+	/**
+	 * Sets the result set client factory.
+	 *
+	 * @param resultSetClientFactory the new result set client factory
+	 */
+	@Required
+	public void setResultSetClientFactory(final ResultSetClientFactory resultSetClientFactory) {
+		this.resultSetClientFactory = resultSetClientFactory;
+	}
+
+}
Index: modules/dnet-hadoop-service/tags/dnet-hadoop-service-2.7.8/src/main/java/eu/dnetlib/data/hadoop/hbase/HBaseDeleteFeeder.java
===================================================================
--- modules/dnet-hadoop-service/tags/dnet-hadoop-service-2.7.8/src/main/java/eu/dnetlib/data/hadoop/hbase/HBaseDeleteFeeder.java	(nonexistent)
+++ modules/dnet-hadoop-service/tags/dnet-hadoop-service-2.7.8/src/main/java/eu/dnetlib/data/hadoop/hbase/HBaseDeleteFeeder.java	(revision 59871)
@@ -0,0 +1,42 @@
+package eu.dnetlib.data.hadoop.hbase;
+
+import java.util.List;
+
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.apache.hadoop.hbase.client.Delete;
+import org.apache.hadoop.hbase.client.Mutation;
+import org.apache.hadoop.hbase.util.Bytes;
+
+import eu.dnetlib.data.transform.Column;
+import eu.dnetlib.data.transform.Row;
+
+/**
+ * The Class HBaseDeleteFeeder performs a batch of Delete operations.
+ */
+public class HBaseDeleteFeeder extends HbaseTableFeeder {
+
+	/**
+	 * Logger.
+	 */
+	private static final Log log = LogFactory.getLog(HBaseDeleteFeeder.class); // NOPMD by marko on 11/24/08 5:02 PM
+
+	/*
+	 * (non-Javadoc)
+	 *
+	 * @see eu.dnetlib.data.hadoop.hbase.HbaseTableFeeder#addOperation(java.util.List, eu.dnetlib.data.transform.Row)
+	 */
+	@Override
+	protected void addOperation(final List<Mutation> buffer, final Row row) {
+		final Delete delete = new Delete(Bytes.toBytes(row.getKey()));
+		delete.setWriteToWAL(true);
+
+		for (final Column<String, byte[]> col : row) {
+			log.debug(String.format("deleting K: '%s' CF:'%s' Q:'%s'", row.getKey(), row.getColumnFamily(), col.getName()));
+			delete.deleteColumns(Bytes.toBytes(row.getColumnFamily()), Bytes.toBytes(col.getName()));
+		}
+
+		buffer.add(delete);
+	}
+
+}
Index: modules/dnet-hadoop-service/tags/dnet-hadoop-service-2.7.8/src/main/java/eu/dnetlib/data/hadoop/hbase/HBasePutFeeder.java
===================================================================
--- modules/dnet-hadoop-service/tags/dnet-hadoop-service-2.7.8/src/main/java/eu/dnetlib/data/hadoop/hbase/HBasePutFeeder.java	(nonexistent)
+++ modules/dnet-hadoop-service/tags/dnet-hadoop-service-2.7.8/src/main/java/eu/dnetlib/data/hadoop/hbase/HBasePutFeeder.java	(revision 59871)
@@ -0,0 +1,42 @@
+package eu.dnetlib.data.hadoop.hbase;
+
+import java.util.List;
+
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.apache.hadoop.hbase.client.Mutation;
+import org.apache.hadoop.hbase.client.Put;
+import org.apache.hadoop.hbase.util.Bytes;
+
+import eu.dnetlib.data.transform.Column;
+import eu.dnetlib.data.transform.Row;
+
+/**
+ * The Class HBasePutFeeder performs a batch of Put operations..
+ */
+public class HBasePutFeeder extends HbaseTableFeeder {
+
+	/**
+	 * Logger.
+	 */
+	private static final Log log = LogFactory.getLog(HBasePutFeeder.class); // NOPMD by marko on 11/24/08 5:02 PM
+
+	/*
+	 * (non-Javadoc)
+	 *
+	 * @see eu.dnetlib.data.hadoop.hbase.HbaseTableFeeder#addOperation(java.util.List, eu.dnetlib.data.transform.Row)
+	 */
+	@Override
+	protected void addOperation(final List<Mutation> buffer, final Row row) {
+		final Put put = new Put(Bytes.toBytes(row.getKey()));
+		put.setWriteToWAL(true);
+
+		for (final Column<String, byte[]> col : row) {
+			log.debug(String.format("adding value to K: '%s' CF:'%s' Q:'%s'", row.getKey(), row.getColumnFamily(), col.getName()));
+			put.add(Bytes.toBytes(row.getColumnFamily()), Bytes.toBytes(col.getName()), col.getValue());
+		}
+
+		buffer.add(put);
+	}
+
+}
Index: modules/dnet-hadoop-service/tags/dnet-hadoop-service-2.7.8/src/main/java/eu/dnetlib/data/hadoop/HadoopServiceCore.java
===================================================================
--- modules/dnet-hadoop-service/tags/dnet-hadoop-service-2.7.8/src/main/java/eu/dnetlib/data/hadoop/HadoopServiceCore.java	(nonexistent)
+++ modules/dnet-hadoop-service/tags/dnet-hadoop-service-2.7.8/src/main/java/eu/dnetlib/data/hadoop/HadoopServiceCore.java	(revision 59871)
@@ -0,0 +1,412 @@
+package eu.dnetlib.data.hadoop;
+
+import java.io.IOException;
+import java.net.URI;
+import java.util.*;
+import java.util.Map.Entry;
+import java.util.stream.Collectors;
+
+import com.google.common.collect.Lists;
+import com.google.common.collect.Maps;
+import com.google.common.collect.Sets;
+import eu.dnetlib.data.hadoop.config.ClusterName;
+import eu.dnetlib.data.hadoop.config.ConfigurationEnumerator;
+import eu.dnetlib.data.hadoop.rmi.HadoopServiceException;
+import eu.dnetlib.data.hadoop.rmi.hbase.Column;
+import eu.dnetlib.data.hadoop.rmi.hbase.HBaseRowDescriptor;
+import eu.dnetlib.data.hadoop.rmi.hbase.schema.HBaseTableDescriptor;
+import eu.dnetlib.data.hadoop.rmi.hbase.schema.HBaseTableRegionInfo;
+import org.apache.commons.lang.StringUtils;
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.FileSystem;
+import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.hbase.HColumnDescriptor;
+import org.apache.hadoop.hbase.HRegionInfo;
+import org.apache.hadoop.hbase.HTableDescriptor;
+import org.apache.hadoop.hbase.client.*;
+import org.apache.hadoop.hbase.util.Bytes;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.beans.factory.annotation.Required;
+
+public class HadoopServiceCore {
+
+	private static final Log log = LogFactory.getLog(HadoopServiceCore.class); // NOPMD by marko on 11/24/08 5:02 PM
+	@Autowired
+	protected ConfigurationEnumerator configurationEnumerator;
+	@Autowired
+	private HadoopClientMap clients;
+	private int maxVersions;
+
+	public List<String> listTables(final ClusterName clusterName) throws IOException, HadoopServiceException {
+		try(final HBaseAdmin admin = clients.getHbaseAdmin(clusterName)) {
+			return Arrays.asList(admin.listTables())
+					.stream()
+					.map(HTableDescriptor::getNameAsString)
+					.collect(Collectors.toList());
+		}
+	}
+
+	public String getHBaseTableDescriptor(final ClusterName clusterName, final String tableName) throws HadoopServiceException, IOException {
+		try(final HBaseAdmin admin = clients.getHbaseAdmin(clusterName)) {
+
+			if (StringUtils.isBlank(tableName)) throw new HadoopServiceException("Table name cannot be empty or null");
+
+			if (admin == null) throw new HadoopServiceException(String.format("HBase admin not available for cluster: '%s'", clusterName.toString()));
+
+			final List<HRegionInfo> tableRegions = admin.getTableRegions(tableName.getBytes());
+
+			final HTableDescriptor desc = admin.getTableDescriptor(tableName.getBytes());
+
+			final Set<String> columns = Sets.newHashSet();
+
+			for (HColumnDescriptor hColDesc : Arrays.asList(desc.getColumnFamilies())) {
+				columns.add(hColDesc.getNameAsString());
+			}
+
+			HBaseTableDescriptor htDescriptor = new HBaseTableDescriptor();
+			htDescriptor.setColumns(columns);
+
+			List<HBaseTableRegionInfo> regions = Lists.newArrayList();
+
+			for (HRegionInfo info : tableRegions) {
+				regions.add(new HBaseTableRegionInfo(new String(info.getStartKey()), new String(info.getEndKey())));
+			}
+			htDescriptor.setRegions(regions);
+
+			if (log.isDebugEnabled()) {
+				log.info("got configuration for table '" + tableName + "': " + htDescriptor.toString());
+			}
+
+			return htDescriptor.toString();
+		}
+	}
+
+	public List<String> describeTable(final ClusterName clusterName, final String table) throws IOException, HadoopServiceException {
+		try(final HBaseAdmin admin = clients.getHbaseAdmin(clusterName)) {
+			final HTableDescriptor desc = admin.getTableDescriptor(table.getBytes());
+			return desc.getFamilies().stream()
+					.map(d -> d.getNameAsString())
+					.collect(Collectors.toList());
+		}
+	}
+
+	public void truncateTable(final ClusterName clusterName, final String table) throws IOException, HadoopServiceException {
+		try(final HBaseAdmin admin = clients.getHbaseAdmin(clusterName)) {
+
+			if (!admin.tableExists(table)) throw new IllegalStateException("cannot truncate unexisting table");
+
+			final HTableDescriptor desc = admin.getTableDescriptor(table.getBytes());
+
+			log.info("disabling table: " + table);
+			admin.disableTable(table);
+
+			log.info("deleting table: " + table);
+			admin.deleteTable(table);
+
+			log.info("creating table: " + table);
+			admin.createTable(desc);
+		}
+	}
+
+	public boolean existTable(final ClusterName clusterName, final String table) throws IOException, HadoopServiceException {
+		try(final HBaseAdmin admin = clients.getHbaseAdmin(clusterName)) {
+
+			return admin.tableExists(table);
+		}
+	}
+
+	public void dropTable(final ClusterName clusterName, final String table) throws IOException, HadoopServiceException {
+		try(final HBaseAdmin admin = clients.getHbaseAdmin(clusterName)) {
+
+			if (!admin.tableExists(table)) throw new IllegalStateException("cannot drop unexisting table: '" + table + "'");
+
+			log.info("disabling table: " + table);
+			admin.disableTable(table);
+
+			log.info("deleting table: " + table);
+			admin.deleteTable(table);
+		}
+	}
+
+	public void createTable(final ClusterName clusterName, final String table, final String tableConfiguration) throws IOException, HadoopServiceException {
+		try(final HBaseAdmin admin = clients.getHbaseAdmin(clusterName)) {
+
+			if (admin.tableExists(table)) throw new IllegalStateException("table already exists");
+
+			if (StringUtils.isBlank(tableConfiguration)) throw new HadoopServiceException("empty table configuration");
+
+			final HBaseTableDescriptor tableConf = HBaseTableDescriptor.fromJSON(tableConfiguration);
+
+			doCreateTable(clusterName, table, tableConf.getColumns(), tableConf.getRegions());
+		}
+	}
+
+	public void createTable(final ClusterName clusterName, final String table, final Set<String> columns) throws IOException, HadoopServiceException {
+		try(final HBaseAdmin admin = clients.getHbaseAdmin(clusterName)) {
+
+			if (admin.tableExists(table)) throw new IllegalStateException("table already exists");
+
+			doCreateTable(clusterName, table, columns, null);
+		}
+	}
+
+	public void doCreateTable(final ClusterName clusterName, final String table, final Set<String> columns, final List<HBaseTableRegionInfo> regions)
+			throws IOException, HadoopServiceException {
+		try(final HBaseAdmin admin = clients.getHbaseAdmin(clusterName)) {
+
+			if (admin.tableExists(table)) throw new IllegalStateException("table already exists");
+
+			final HTableDescriptor desc = new HTableDescriptor(table);
+			for (final String column : columns) {
+				final HColumnDescriptor hds = new HColumnDescriptor(column);
+				hds.setMaxVersions(getMaxVersions());
+				desc.addFamily(hds);
+			}
+
+			log.info("creating hbase table: " + table);
+
+			if (regions != null && !regions.isEmpty()) {
+				log.debug(String.format("create using %s regions: %s", regions.size(), regions));
+				admin.createTable(desc, getSplitKeys(regions));
+			} else {
+				admin.createTable(desc);
+			}
+
+			log.info("created hbase table: [" + table + "]");
+			log.debug("descriptor: [" + desc.toString() + "]");
+		}
+	}
+
+	private byte[][] getSplitKeys(final List<HBaseTableRegionInfo> regions) {
+		byte[][] splits = new byte[regions.size() - 1][];
+		for (int i = 0; i < regions.size() - 1; i++) {
+			splits[i] = regions.get(i).getEndKey().getBytes();
+		}
+		return splits;
+	}
+
+	public void ensureTable(final ClusterName clusterName, final String table, final Set<String> columns) throws IOException, HadoopServiceException {
+		try(final HBaseAdmin admin = clients.getHbaseAdmin(clusterName)) {
+
+			if (!admin.tableExists(table)) {
+				createTable(clusterName, table, columns);
+			} else {
+				final HTableDescriptor desc = admin.getTableDescriptor(Bytes.toBytes(table));
+
+				final Set<String> foundColumns = desc.getFamilies().stream()
+						.map(d -> d.getNameAsString())
+						.collect(Collectors.toCollection(HashSet::new));
+
+				log.info("ensuring columns on table " + table + ": " + columns);
+				final Collection<String> missingColumns = Sets.difference(columns, foundColumns);
+				if (!missingColumns.isEmpty()) {
+
+					if (admin.isTableEnabled(table)) {
+						admin.disableTable(table);
+					}
+
+					for (final String column : missingColumns) {
+						log.info("hbase table: '" + table + "', adding column: " + column);
+						admin.addColumn(table, new HColumnDescriptor(column));
+					}
+
+					admin.enableTable(table);
+				}
+			}
+		}
+	}
+
+	public void writeOnHBase(final ClusterName clusterName, final String tableName, final List<Put> puts) throws IOException {
+		final Configuration conf = configurationEnumerator.get(clusterName);
+		final HTable table = new HTable(conf, tableName);
+
+		try {
+			table.put(puts);
+		} finally {
+			table.flushCommits();
+			table.close();
+		}
+	}
+
+	public void deleteFromHBase(final ClusterName clusterName, final String tableName, final List<Delete> deletes) throws IOException {
+		final Configuration conf = configurationEnumerator.get(clusterName);
+		final HTable table = new HTable(conf, tableName);
+		try {
+			table.delete(deletes);
+		} finally {
+			table.flushCommits();
+			table.close();
+		}
+	}
+
+	public void deleteColumnsFromHBase(final ClusterName clusterName, final String tableName, final List<HBaseRowDescriptor> columns) throws IOException {
+		final Configuration conf = configurationEnumerator.get(clusterName);
+		final HTable table = new HTable(conf, tableName);
+		try {
+			for(HBaseRowDescriptor desc : columns) {
+
+				final Delete d = new Delete(Bytes.toBytes(desc.getRowKey()));
+				d.setWriteToWAL(true);
+				for(Column c : desc.getColumns()) {
+					for(String qualifier : c.getQualifier()) {
+						log.info(String.format("delete from row '%s' cf '%s:%s'", desc.getRowKey(), c.getFamily(), qualifier));
+						d.deleteColumns(Bytes.toBytes(c.getFamily()), Bytes.toBytes(qualifier));
+					}
+				}
+				table.delete(d);
+			}
+		} finally {
+			table.flushCommits();
+			table.close();
+		}
+	}
+
+	public Result getRow(final ClusterName clusterName, final String tableName, final byte[] id) throws IOException {
+		final Configuration conf = configurationEnumerator.get(clusterName);
+		final HTable table = new HTable(conf, tableName);
+		try {
+			return table.get(new Get(id));
+		} finally {
+			table.close();
+		}
+	}
+
+	public Map<String, HBaseRowDescriptor> describeRows(final ClusterName clusterName, final String tableName, final List<String> rowKeys) throws IOException {
+		final Map<String, HBaseRowDescriptor> map = Maps.newHashMap();
+		for(String rowKey : rowKeys) {
+			map.put(rowKey, describeRow(clusterName, tableName, rowKey));
+		}
+		return map;
+	}
+
+	public HBaseRowDescriptor describeRow(final ClusterName clusterName, final String tableName, final String rowKey) throws IOException {
+		final Configuration conf = configurationEnumerator.get(clusterName);
+		final HTable table = new HTable(conf, tableName);
+
+		final HBaseRowDescriptor desc = new HBaseRowDescriptor();
+
+		try {
+			final Result r = table.get(new Get(Bytes.toBytes(rowKey)));
+
+			if (r.isEmpty()) {
+				return desc;
+			}
+
+			final List<Column> columns = Lists.newArrayList();
+
+			for(Entry<byte[], NavigableMap<byte[], byte[]>> e : r.getNoVersionMap().entrySet()) {
+				final Set<byte[]> qualifiers = e.getValue().keySet();
+				final String family = new String(e.getKey());
+				final Column col = new Column(family);
+
+				for(byte[] q : qualifiers) {
+					String qs = new String(q);
+					col.getQualifier().add(qs);
+				}
+				columns.add(col);
+			}
+			desc.setColumns(columns);
+			desc.setRowKey(rowKey);
+
+			return desc;
+		} finally {
+			table.close();
+		}
+	}
+
+	public List<Result> getRows(final ClusterName clusterName, final String tableName, final Scan scan) throws IOException {
+		final Configuration conf = configurationEnumerator.get(clusterName);
+		try(final HTable table = new HTable(conf, tableName)) {
+			final ResultScanner rs = table.getScanner(scan);
+			try {
+				return Lists.newArrayList(rs.iterator());
+			} finally {
+				rs.close();
+			}
+		}
+	}
+
+	public boolean deleteFromHdfs(final ClusterName clusterName, final String path) throws HadoopServiceException {
+		if (StringUtils.isBlank(path))
+			throw new HadoopServiceException("Cannot deleteFromHBase an empty HDFS path.");
+
+		final Configuration conf = configurationEnumerator.get(clusterName);
+
+		try(final FileSystem hdfs = FileSystem.get(conf)) {
+			final Path absolutePath = new Path(URI.create(conf.get("fs.defaultFS") + path));
+
+			if (hdfs.exists(absolutePath)) {
+				log.debug("deleteFromHBase path: " + absolutePath.toString());
+				hdfs.delete(absolutePath, true);
+				log.info("deleted path: " + absolutePath.toString());
+				return true;
+			} else {
+				log.warn("cannot deleteFromHBase unexisting path: " + absolutePath.toString());
+				return false;
+			}
+		} catch (IOException e) {
+			throw new HadoopServiceException(e);
+		}
+	}
+
+	public boolean createHdfsDir(final ClusterName clusterName, final String path, final boolean force) throws HadoopServiceException {
+		if (StringUtils.isBlank(path))
+			throw new HadoopServiceException("Cannot create an empty HDFS path.");
+
+		final Configuration conf = configurationEnumerator.get(clusterName);
+
+		try(final FileSystem hdfs = FileSystem.get(conf)) {
+			final Path absolutePath = new Path(URI.create(conf.get("fs.defaultFS") + path));
+			if (!hdfs.exists(absolutePath)) {
+				hdfs.mkdirs(absolutePath);
+				log.info("created path: " + absolutePath.toString());
+				return true;
+			} else if (force) {
+				log.info(String.format("found directory '%s', force delete it", absolutePath.toString()));
+				hdfs.delete(absolutePath, true);
+
+				hdfs.mkdirs(absolutePath);
+				log.info("created path: " + absolutePath.toString());
+				return true;
+			} else {
+				log.info(String.format("directory already exists: '%s', nothing to do", absolutePath.toString()));
+				return false;
+			}
+		} catch (IOException e) {
+			throw new HadoopServiceException(e);
+		}
+	}
+
+	public boolean existHdfsPath(final ClusterName clusterName, final String path) throws HadoopServiceException {
+		if (StringUtils.isBlank(path))
+			throw new HadoopServiceException("invalid empty path");
+
+		final Configuration conf = configurationEnumerator.get(clusterName);
+		try(final FileSystem hdfs = FileSystem.get(conf)) {
+			final Path absolutePath = new Path(URI.create(conf.get("fs.defaultFS") + path));
+			return hdfs.exists(absolutePath);
+		} catch (IOException e) {
+			throw new HadoopServiceException(e);
+		}
+	}
+
+	public Configuration getClusterConiguration(final ClusterName clusterName) {
+		return configurationEnumerator.get(clusterName);
+	}
+
+	public int getMaxVersions() {
+		return maxVersions;
+	}
+
+	@Required
+	public void setMaxVersions(final int maxVersions) {
+		this.maxVersions = maxVersions;
+	}
+
+	public HadoopClientMap getClients() {
+		return clients;
+	}
+
+}
Index: modules/dnet-hadoop-service/tags/dnet-hadoop-service-2.7.8/src/main/java/eu/dnetlib/data/hadoop/mapred/JobClientFactory.java
===================================================================
--- modules/dnet-hadoop-service/tags/dnet-hadoop-service-2.7.8/src/main/java/eu/dnetlib/data/hadoop/mapred/JobClientFactory.java	(nonexistent)
+++ modules/dnet-hadoop-service/tags/dnet-hadoop-service-2.7.8/src/main/java/eu/dnetlib/data/hadoop/mapred/JobClientFactory.java	(revision 59871)
@@ -0,0 +1,28 @@
+package eu.dnetlib.data.hadoop.mapred;
+
+import java.io.IOException;
+
+import eu.dnetlib.data.hadoop.AbstractHadoopClient;
+import eu.dnetlib.data.hadoop.config.ClusterName;
+import eu.dnetlib.data.hadoop.rmi.HadoopServiceException;
+import org.apache.hadoop.mapred.JobClient;
+
+/**
+ * Factory bean for jobClient instances
+ *
+ * @author claudio
+ *
+ */
+public class JobClientFactory extends AbstractHadoopClient {
+
+	public JobClient newInstance(final ClusterName clusterName, final String userName) throws IOException {
+		setHadoopUser(userName);
+		return new JobClient(configurationEnumerator.get(clusterName));
+	}
+
+	public JobClient newInstance(final ClusterName clusterName) throws IOException, HadoopServiceException {
+		setHadoopUser();
+		return new JobClient(configurationEnumerator.get(clusterName));
+	}
+
+}
Index: modules/dnet-hadoop-service/tags/dnet-hadoop-service-2.7.8/src/main/java/eu/dnetlib/data/hadoop/mapred/MapreduceJobMonitor.java
===================================================================
--- modules/dnet-hadoop-service/tags/dnet-hadoop-service-2.7.8/src/main/java/eu/dnetlib/data/hadoop/mapred/MapreduceJobMonitor.java	(nonexistent)
+++ modules/dnet-hadoop-service/tags/dnet-hadoop-service-2.7.8/src/main/java/eu/dnetlib/data/hadoop/mapred/MapreduceJobMonitor.java	(revision 59871)
@@ -0,0 +1,146 @@
+package eu.dnetlib.data.hadoop.mapred;
+
+import java.io.IOException;
+import java.util.Date;
+import java.util.Map;
+
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.apache.hadoop.mapred.Counters;
+import org.apache.hadoop.mapred.Counters.Counter;
+import org.apache.hadoop.mapred.Counters.Group;
+import org.apache.hadoop.mapred.JobClient;
+import org.apache.hadoop.mapred.JobStatus;
+import org.apache.hadoop.mapred.RunningJob;
+
+import com.google.common.collect.Maps;
+
+import eu.dnetlib.data.hadoop.action.JobCompletion;
+import eu.dnetlib.data.hadoop.action.JobMonitor;
+import eu.dnetlib.data.hadoop.rmi.HadoopServiceException;
+
+public class MapreduceJobMonitor extends JobMonitor {
+
+	private static final Log log = LogFactory.getLog(MapreduceJobMonitor.class); // NOPMD by marko on 11/24/08 5:02 PM
+
+	private final RunningJob runningJob;
+
+	private final JobClient jobClient;
+
+	public MapreduceJobMonitor(final RunningJob runningJob, final JobCompletion callback) {
+		this(null, runningJob, callback);
+	}
+
+	public MapreduceJobMonitor(final JobClient jobClient, final RunningJob runningJob, final JobCompletion callback) {
+		super(callback);
+		this.runningJob = runningJob;
+		this.jobClient = jobClient;
+	}
+
+	@Override
+	public void run() {
+		try {
+			log.info("waiting for job completion: " + getRunningJob().getID().getId());
+
+			int runState = getRunState();
+			while (!getRunningJob().isComplete()) {
+				Thread.sleep(monitorSleepTimeSecs * 1000);
+
+				// if the run state has changed we update the last activity
+				final int currentState = getRunState();
+				if (currentState != runState) {
+					runState = currentState;
+					lastActivity = new Date(System.currentTimeMillis());
+				}
+			}
+
+			if (getRunningJob().isSuccessful()) {
+				log.info("job successful: " + getRunningJob().getID().getId());
+				getCallback().done(asMap(getRunningJob().getCounters()));
+			} else {
+				final String msg = "hadoop job: " + getHadoopId() + " failed with status: " + getStatus();
+				getCallback().failed(msg, new HadoopServiceException(msg));
+			}
+		} catch (final Throwable e) {
+			getCallback().failed(getHadoopId(), e);
+		} finally {
+			try {
+				if (jobClient != null) {
+					jobClient.close();
+				}
+			} catch (IOException e) {
+				throw new RuntimeException("unable to close jobClient", e);
+			}
+		}
+	}
+
+	@Override
+	public String getHadoopId() {
+		return String.valueOf(getRunningJob().getID().getId());
+	}
+
+	// TODO: many hadoop classes are available under the newer package org.apache.hadoop.mapreduce.Counters,
+	// but the JobClient and its returned objects are still using the old and deprecated org.apache.hadoop.mapred
+	protected Map<String, String> asMap(final Counters counters) {
+		final Map<String, String> res = Maps.newHashMap();
+		if (counters != null) {
+			for (final Group group : counters) {
+				for (final Counter counter : group) {
+					res.put(group.getDisplayName() + "." + counter.getDisplayName(), String.valueOf(counter.getValue()));
+				}
+			}
+		}
+		return res;
+	}
+
+	public RunningJob getRunningJob() {
+		return runningJob;
+	}
+
+	@Override
+	public String getStatus() {
+		try {
+			return JobStatus.getJobRunState(getRunState());
+		} catch (final IOException e) {
+			log.error("error accessing job status", e);
+			return "UNKNOWN";
+		}
+	}
+
+	private int getRunState() throws IOException {
+		return getRunningJob().getJobStatus().getRunState();
+	}
+
+	@Override
+	public Date getLastActivity() {
+		return lastActivity;
+	}
+
+	@Override
+	public Date getStartTime() throws HadoopServiceException {
+		try {
+			return new Date(getRunningJob().getJobStatus().getStartTime());
+		} catch (final IOException e) {
+			throw new HadoopServiceException("unable to read job start time", e);
+		}
+	}
+
+	@Override
+	public String getTrackerUrl() {
+		return getRunningJob().getTrackingURL();
+	}
+
+	@Override
+	public void kill() {
+		try {
+			log.info("killing job: " + getHadoopId());
+			getRunningJob().killJob();
+		} catch (final IOException e) {
+			log.error("unable to kill job: " + getHadoopId(), e);
+		}
+	}
+
+	public JobClient getJobClient() {
+		return jobClient;
+	}
+}
Index: modules/dnet-hadoop-service/tags/dnet-hadoop-service-2.7.8/src/main/java/eu/dnetlib/data/hadoop/action/SubmitMapreduceJobAction.java
===================================================================
--- modules/dnet-hadoop-service/tags/dnet-hadoop-service-2.7.8/src/main/java/eu/dnetlib/data/hadoop/action/SubmitMapreduceJobAction.java	(nonexistent)
+++ modules/dnet-hadoop-service/tags/dnet-hadoop-service-2.7.8/src/main/java/eu/dnetlib/data/hadoop/action/SubmitMapreduceJobAction.java	(revision 59871)
@@ -0,0 +1,118 @@
+package eu.dnetlib.data.hadoop.action;
+
+import java.io.IOException;
+import java.io.StringWriter;
+import java.util.Map;
+import java.util.Map.Entry;
+
+import eu.dnetlib.data.hadoop.HadoopJob;
+import eu.dnetlib.data.hadoop.config.ClusterName;
+import eu.dnetlib.data.hadoop.mapred.MapreduceJobMonitor;
+import eu.dnetlib.data.hadoop.rmi.HadoopServiceException;
+import eu.dnetlib.data.hadoop.utils.JobProfile;
+import eu.dnetlib.data.hadoop.utils.ScanFactory;
+import eu.dnetlib.data.hadoop.utils.ScanProperties;
+import eu.dnetlib.enabling.tools.blackboard.BlackboardJob;
+import eu.dnetlib.miscutils.functional.xml.IndentXmlString;
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.mapred.JobClient;
+import org.apache.hadoop.mapred.JobConf;
+import org.apache.hadoop.mapred.RunningJob;
+
+public class SubmitMapreduceJobAction extends AbstractSubmitAction {
+
+	/**
+	 * logger.
+	 */
+	private static final Log log = LogFactory.getLog(SubmitMapreduceJobAction.class); // NOPMD by marko on 11/24/08 5:02 PM
+
+	@Override
+	public void submit(final JobCompletion callback, final BlackboardJob bbJob, final String jobName, final JobProfile jobProfile)
+			throws HadoopServiceException {
+
+		final ClusterName clusterName = ClusterName.valueOf(bbJob.getParameters().get("cluster"));
+
+		try {
+			final JobConf jobConf = prepareJob(getConf(clusterName), jobName, jobProfile, bbJob.getParameters());
+
+			logJobDetails(jobConf);
+
+			final JobClient jtClient = hadoopClientMap.getJtClient(clusterName);
+			final RunningJob runningJob = jtClient.submitJob(jobConf);
+
+			final String jobId = newJobId(clusterName, runningJob.getID().getId());
+
+			jobRegistry.registerJob(
+					HadoopJob.newInstance(jobId, clusterName, jobProfile, new MapreduceJobMonitor(jtClient, runningJob, callback)));
+
+		} catch (final IOException e) {
+			throw new HadoopServiceException("error executing hadoop job: " + jobName, e);
+		}
+	}
+
+	protected JobConf prepareJob(final Configuration configuration, final String jobName, final JobProfile jobProfile, final Map<String, String> parameters)
+			throws IOException, HadoopServiceException {
+
+		log.info("creating job: " + jobName);
+
+		final JobConf jobConf = new JobConf(configuration);
+		jobConf.setJobName(jobName);
+		jobConf.set("dnet.mapred.job.description", jobProfile.getDescription());
+
+		final String jobLib = getJobLib(configuration, jobProfile);
+		jobConf.setJar(new Path(jobLib).toString());
+
+		set(jobConf, jobProfile.getJobDefinition());
+		set(jobConf, parameters);
+
+		// if we're reading from hbase table then we need also to set a scanner.
+		final ScanProperties scanProperties = jobProfile.getScanProperties();
+		if (jobProfile.getRequiredParams().contains("hbase.mapreduce.inputtable") && (scanProperties != null)) {
+			jobConf.set("hbase.mapreduce.scan", ScanFactory.getScan(scanProperties));
+		}
+
+		return jobConf;
+	}
+
+	protected String getJobLib(final Configuration configuration, final JobProfile jobProfile) throws HadoopServiceException {
+		String jobLib = getDefaultLibPath(configuration.get("fs.defaultFS"));
+
+		if (jobProfile.getJobDefinition().containsKey("job.lib")) {
+			jobLib = jobProfile.getJobDefinition().get("job.lib");
+		}
+
+		if ((jobLib == null) || jobLib.isEmpty()) throw new HadoopServiceException("job.lib must refer to an absolute or relative HDFS path");
+		if (!jobLib.startsWith("hdfs://")) {
+			jobLib = configuration.get("fs.defaultFS") + jobLib;
+		}
+
+		log.info("using job.lib: " + jobLib);
+		return jobLib;
+	}
+
+	protected void set(final JobConf jobConf, final Map<String, String> properties) {
+		for (final Entry<String, String> e : properties.entrySet()) {
+			if (checkHdfsProperty(e)) {
+				final String v = jobConf.get("fs.defaultFS") + e.getValue();
+				e.setValue(v);
+			}
+			jobConf.set(e.getKey(), e.getValue());
+		}
+	}
+
+	protected void logJobDetails(final JobConf jobConf) {
+		final StringWriter sw = new StringWriter();
+		try {
+			jobConf.writeXml(sw);
+			if (log.isDebugEnabled()) {
+				log.debug("\n" + IndentXmlString.apply(sw.toString()));
+			}
+		} catch (final IOException e) {
+			log.warn("unable to log job details: " + jobConf.getJobName());
+		}
+	}
+
+}
Index: modules/dnet-hadoop-service/tags/dnet-hadoop-service-2.7.8/src/main/java/eu/dnetlib/data/hadoop/action/AbstractSubmitAction.java
===================================================================
--- modules/dnet-hadoop-service/tags/dnet-hadoop-service-2.7.8/src/main/java/eu/dnetlib/data/hadoop/action/AbstractSubmitAction.java	(nonexistent)
+++ modules/dnet-hadoop-service/tags/dnet-hadoop-service-2.7.8/src/main/java/eu/dnetlib/data/hadoop/action/AbstractSubmitAction.java	(revision 59871)
@@ -0,0 +1,176 @@
+package eu.dnetlib.data.hadoop.action;
+
+import java.net.URI;
+import java.util.List;
+import java.util.Map;
+
+import org.apache.commons.lang.StringUtils;
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.apache.hadoop.fs.Path;
+import org.springframework.beans.factory.annotation.Autowired;
+
+import com.google.common.collect.Iterables;
+import com.google.common.collect.Sets;
+
+import eu.dnetlib.data.hadoop.HadoopClientMap;
+import eu.dnetlib.data.hadoop.JobRegistry;
+import eu.dnetlib.data.hadoop.config.ClusterName;
+import eu.dnetlib.data.hadoop.rmi.HadoopServiceException;
+import eu.dnetlib.data.hadoop.utils.JobProfile;
+import eu.dnetlib.enabling.is.lookup.rmi.ISLookUpException;
+import eu.dnetlib.enabling.is.lookup.rmi.ISLookUpService;
+import eu.dnetlib.enabling.is.registry.rmi.ISRegistryException;
+import eu.dnetlib.enabling.is.registry.rmi.ISRegistryService;
+import eu.dnetlib.enabling.tools.blackboard.BlackboardJob;
+import eu.dnetlib.enabling.tools.blackboard.BlackboardServerHandler;
+import eu.dnetlib.miscutils.datetime.DateUtils;
+import eu.dnetlib.miscutils.functional.hash.Hashing;
+
+public abstract class AbstractSubmitAction extends AbstractHadoopAction {
+
+	private static final Log log = LogFactory.getLog(AbstractSubmitAction.class); // NOPMD by marko on 11/24/08 5:02 PM
+
+	@Autowired
+	protected HadoopClientMap hadoopClientMap;
+
+	@Autowired
+	protected JobRegistry jobRegistry;
+
+	protected abstract void submit(final JobCompletion callback, final BlackboardJob job, final String jobName, final JobProfile jobProfile)
+			throws HadoopServiceException;
+
+	@Override
+	protected void executeAsync(final BlackboardServerHandler handler, final BlackboardJob job) throws HadoopServiceException {
+		final String jobName = job.getParameters().get("job.name");
+		final JobProfile jobProfile = loadISJobConfiguration(jobName, job.getParameters());
+
+		validateJobParams(handler, job, jobName, jobProfile);
+
+		if (!isSimulation(job)) {
+
+			submit(newCompletionCallback(handler, job), job, jobName, jobProfile);
+			updateJobStatus(jobName);
+		} else {
+			log.info(String.format("simulating job: '%s', done!", jobName));
+			handler.done(job);
+		}
+	}
+
+	private boolean isSimulation(final BlackboardJob job) {
+		if (!job.getParameters().containsKey("simulation")) { return false; }
+		final String sim = job.getParameters().get("simulation");
+		return !StringUtils.isBlank(sim) && sim.equals("true");
+	}
+
+	protected void updateJobStatus(final String jobName) {
+		incrementCumulativeRun(jobName);
+		incrementRunningJobs(jobName);
+		updateDate(jobName);
+	}
+
+	protected void incrementRunningJobs(final String jobName) {
+		log.debug("increment #running jobs: " + jobName);
+		updateCountElement(jobName, "RUNNING_INSTANCES", "+ 1");
+	}
+
+	protected void decrementRunningJobs(final String jobName) {
+		log.debug("decrement #running jobs: " + jobName);
+		updateCountElement(jobName, "RUNNING_INSTANCES", "- 1");
+	}
+
+	protected void incrementCumulativeRun(final String jobName) {
+		log.debug("increment #cumulative runs for job: " + jobName);
+		updateCountElement(jobName, "CUMULATIVE_RUN", "+ 1");
+	}
+
+	protected void updateDate(final String jobName) {
+		log.info("increment last submission date for job: " + jobName);
+		final String xquery =
+				"for $x in collection('')/RESOURCE_PROFILE[" + ".//RESOURCE_TYPE/@value='HadoopJobConfigurationDSResourceType' and .//HADOOP_JOB/@name='"
+						+ jobName + "'] " + " return update value $x//LAST_SUBMISSION_DATE/@value with '" + DateUtils.now_ISO8601() + "' ";
+
+		executeXUpdate(xquery);
+	}
+
+	private void updateCountElement(final String jobName, final String element, final String delta) {
+		final String xquery =
+				"let $x := //RESOURCE_PROFILE[" + ".//RESOURCE_TYPE/@value='HadoopJobConfigurationDSResourceType' and .//HADOOP_JOB/@name='" + jobName
+						+ "'], $tot := $x//STATUS/" + element + "/@value/number() " + delta + " return update replace $x//STATUS/" + element + " with <"
+						+ element + " value='{$tot}' />";
+
+		executeXUpdate(xquery);
+	}
+
+	protected String newJobId(final ClusterName clusterName, final Object internalId) {
+		return "job-" + Hashing.md5(clusterName.toString() + internalId.toString());
+	}
+
+	protected String getDefaultLibPath(final String defaultFs) throws HadoopServiceException {
+		try {
+			final String libPath = queryForServiceProperty("defaultLibPath");
+			final Path path = new Path(URI.create(defaultFs + libPath));
+
+			return path.toString();
+		} catch (ISLookUpException e) {
+			throw new HadoopServiceException("unable to get default lib path", e);
+		}
+	}
+
+	protected void validateJobParams(final BlackboardServerHandler handler, final BlackboardJob bbJob, final String jobName, final JobProfile jobProfile)
+			throws HadoopServiceException {
+		if (!bbJob.getParameters().keySet().containsAll(jobProfile.getRequiredParams())) {
+			String msg =
+					"required parameter is missing for job: " + jobName + ", required params: " + jobProfile.getRequiredParams() + "\n\nmissing params: "
+							+ Sets.difference(jobProfile.getRequiredParams(), bbJob.getParameters().keySet());
+			log.error(msg);
+			HadoopServiceException e = new HadoopServiceException(msg);
+			handler.failed(bbJob, e);
+			throw e;
+		}
+	}
+
+	protected JobCompletion newCompletionCallback(final BlackboardServerHandler handler, final BlackboardJob bbJob) {
+		final String jobName = bbJob.getParameters().get("job.name");
+		return new JobCompletion() {
+
+			@Override
+			public void done(final Map<String, String> properties) {
+				bbJob.getParameters().putAll(properties);
+				log.info(String.format("%s completed successfully, returning %s output params", jobName, properties.size()));
+				handler.done(bbJob);
+				decrementRunningJobs(jobName);
+			}
+
+			@Override
+			public void failed(final String msg, final Throwable e) {
+				log.debug(msg);
+				handler.failed(bbJob, e);
+				decrementRunningJobs(jobName);
+			}
+		};
+	}
+
+	private String queryForServiceProperty(final String key) throws ISLookUpException {
+		return getServiceConfigValue("for $x in /RESOURCE_PROFILE[.//RESOURCE_TYPE/@value='HadoopServiceResourceType'] return $x//SERVICE_PROPERTIES/PROPERTY[./@ key='"
+				+ key + "']/@value/string()");
+	}
+
+	private String getServiceConfigValue(final String xquery) throws ISLookUpException {
+		log.debug("quering for service property: " + xquery);
+		final List<String> urls = serviceLocator.getService(ISLookUpService.class).quickSearchProfile(xquery);
+		if (urls == null || urls.size() != 1) { throw new IllegalStateException("unable to find unique service property, xquery: " + xquery); }
+		return Iterables.getOnlyElement(urls);
+	}
+
+	private boolean executeXUpdate(final String xupdate) {
+		try {
+			log.debug("running xupdate: " + xupdate);
+			return serviceLocator.getService(ISRegistryService.class).executeXUpdate(xupdate);
+		} catch (ISRegistryException e) {
+			log.error("unable to run xupdate: " + xupdate, e);
+			return false;
+		}
+	}
+
+}
Index: modules/dnet-hadoop-service/tags/dnet-hadoop-service-2.7.8/src/main/java/eu/dnetlib/data/hadoop/action/AbstractHadoopAction.java
===================================================================
--- modules/dnet-hadoop-service/tags/dnet-hadoop-service-2.7.8/src/main/java/eu/dnetlib/data/hadoop/action/AbstractHadoopAction.java	(nonexistent)
+++ modules/dnet-hadoop-service/tags/dnet-hadoop-service-2.7.8/src/main/java/eu/dnetlib/data/hadoop/action/AbstractHadoopAction.java	(revision 59871)
@@ -0,0 +1,137 @@
+package eu.dnetlib.data.hadoop.action;
+
+import java.io.StringReader;
+import java.util.Map;
+import java.util.Map.Entry;
+import java.util.Set;
+import java.util.concurrent.Executor;
+import java.util.concurrent.Executors;
+
+import javax.annotation.Resource;
+
+import org.apache.commons.lang.exception.ExceptionUtils;
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.apache.hadoop.conf.Configuration;
+import org.dom4j.Document;
+import org.dom4j.DocumentException;
+import org.dom4j.Node;
+import org.dom4j.io.SAXReader;
+import org.springframework.beans.factory.annotation.Autowired;
+
+import com.google.common.collect.Sets;
+
+import eu.dnetlib.data.hadoop.config.ClusterName;
+import eu.dnetlib.data.hadoop.config.ConfigurationEnumerator;
+import eu.dnetlib.data.hadoop.rmi.HadoopBlackboardActions;
+import eu.dnetlib.data.hadoop.rmi.HadoopJobType;
+import eu.dnetlib.data.hadoop.rmi.HadoopServiceException;
+import eu.dnetlib.data.hadoop.utils.JobProfile;
+import eu.dnetlib.data.hadoop.utils.ScanFactory;
+import eu.dnetlib.enabling.is.lookup.rmi.ISLookUpDocumentNotFoundException;
+import eu.dnetlib.enabling.is.lookup.rmi.ISLookUpException;
+import eu.dnetlib.enabling.is.lookup.rmi.ISLookUpService;
+import eu.dnetlib.enabling.locators.UniqueServiceLocator;
+import eu.dnetlib.enabling.tools.blackboard.BlackboardJob;
+import eu.dnetlib.enabling.tools.blackboard.BlackboardServerAction;
+import eu.dnetlib.enabling.tools.blackboard.BlackboardServerHandler;
+
+public abstract class AbstractHadoopAction implements BlackboardServerAction<HadoopBlackboardActions> {
+
+	private static final Log log = LogFactory.getLog(AbstractHadoopAction.class); // NOPMD by marko on 11/24/08 5:02 PM
+
+	/**
+	 * Special I/O hdfs property names for which we support relative paths.
+	 */
+	public final static Set<String> HDFS_SPECIAL_PROPERTIES = Sets.newHashSet("mapred.input.dir", "mapred.output.dir");
+
+	@Resource
+	protected UniqueServiceLocator serviceLocator;
+
+	@Autowired
+	protected ConfigurationEnumerator configurationEnumerator;
+
+	private final Executor executor = Executors.newCachedThreadPool();
+
+	protected abstract void executeAsync(final BlackboardServerHandler handler, final BlackboardJob job) throws HadoopServiceException;
+
+	@Override
+	public void execute(final BlackboardServerHandler handler, final BlackboardJob job) {
+		executor.execute(() -> {
+			try {
+				handler.ongoing(job);
+				executeAsync(handler, job);
+			} catch (final Throwable e) {
+				log.error(ExceptionUtils.getFullStackTrace(e));
+				handler.failed(job, e);
+			}
+		});
+	}
+
+	protected JobProfile loadISJobConfiguration(final String jobName, final Map<String, String> bbParams) throws HadoopServiceException {
+
+		log.info("reading job configuration profile: " + jobName);
+
+		try {
+			final String profile = serviceLocator.getService(ISLookUpService.class).getResourceProfileByQuery(
+					"/RESOURCE_PROFILE[.//RESOURCE_TYPE/@value = 'HadoopJobConfigurationDSResourceType' and .//HADOOP_JOB/@name='" + jobName + "']");
+			return parseJobProfile(profile, bbParams);
+		} catch (final ISLookUpDocumentNotFoundException e) {
+			throw new HadoopServiceException("cannot find job profile: " + jobName, e);
+		} catch (final ISLookUpException e) {
+			throw new HadoopServiceException("unable to read job profile: " + jobName, e);
+		}
+	}
+
+	private JobProfile parseJobProfile(final String profile, final Map<String, String> bbParams) throws HadoopServiceException {
+		final JobProfile jobProfile = new JobProfile();
+		try {
+			final Document doc = new SAXReader().read(new StringReader(profile));
+
+			log.debug("setting job description");
+			jobProfile.setDescription(doc.valueOf("//DESCRIPTION"));
+
+			log.debug("setting job name");
+			jobProfile.setName(doc.valueOf("//HADOOP_JOB/@name"));
+
+			log.debug("setting job type");
+			jobProfile.setJobType(HadoopJobType.valueOf(doc.valueOf("//HADOOP_JOB/@type")));
+
+			log.debug("setting job static configuration");
+			for (final Object o : doc.selectNodes("//STATIC_CONFIGURATION/PROPERTY")) {
+				final Node node = (Node) o;
+				jobProfile.getJobDefinition().put(node.valueOf("./@key"), node.valueOf("./@value"));
+			}
+
+			log.debug("setting job required parameters");
+			for (final Object o : doc.selectNodes("//JOB_INTERFACE/PARAM[./@required = 'true']")) {
+				final Node node = (Node) o;
+				jobProfile.getRequiredParams().add(node.valueOf("./@name"));
+			}
+
+			if (doc.selectSingleNode("//SCAN") != null) {
+				jobProfile.setScanProperties(ScanFactory.parseScanProperties(doc, bbParams));
+			}
+
+		} catch (final DocumentException e) {
+			throw new HadoopServiceException("cannot parse job profile");
+		}
+
+		if (jobProfile.isEmpty()) throw new HadoopServiceException("job configuration is empty");
+
+		return jobProfile;
+	}
+
+	protected Configuration getConf(final ClusterName clusterName) {
+		return configurationEnumerator.get(clusterName);
+	}
+
+	protected boolean checkHdfsProperty(final Entry<String, String> e) {
+		return HDFS_SPECIAL_PROPERTIES.contains(e.getKey()) && !e.getValue().isEmpty() && !e.getValue().startsWith("hdfs://");
+	}
+
+	public UniqueServiceLocator getServiceLocator() {
+		return serviceLocator;
+	}
+
+}
Index: modules/dnet-hadoop-service/tags/dnet-hadoop-service-2.7.8/src/main/java/eu/dnetlib/data/hadoop/action/SubmitOozieJobAction.java
===================================================================
--- modules/dnet-hadoop-service/tags/dnet-hadoop-service-2.7.8/src/main/java/eu/dnetlib/data/hadoop/action/SubmitOozieJobAction.java	(nonexistent)
+++ modules/dnet-hadoop-service/tags/dnet-hadoop-service-2.7.8/src/main/java/eu/dnetlib/data/hadoop/action/SubmitOozieJobAction.java	(revision 59871)
@@ -0,0 +1,96 @@
+package eu.dnetlib.data.hadoop.action;
+
+import java.util.Map;
+import java.util.Map.Entry;
+import java.util.Properties;
+import java.util.Set;
+
+import com.google.common.base.Splitter;
+import com.google.common.collect.Sets;
+import org.apache.commons.lang.StringUtils;
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.oozie.client.OozieClient;
+import org.apache.oozie.client.OozieClientException;
+
+import eu.dnetlib.data.hadoop.HadoopJob;
+import eu.dnetlib.data.hadoop.config.ClusterName;
+import eu.dnetlib.data.hadoop.oozie.OozieJobMonitor;
+import eu.dnetlib.data.hadoop.rmi.HadoopServiceException;
+import eu.dnetlib.data.hadoop.utils.JobProfile;
+import eu.dnetlib.enabling.tools.blackboard.BlackboardJob;
+
+public class SubmitOozieJobAction extends AbstractSubmitAction {
+
+	private static final Log log = LogFactory.getLog(SubmitOozieJobAction.class); // NOPMD by marko on 11/24/08 5:02 PM
+
+	public static final String OOZIE_REPORT_ACTIONS = "oozie.report.actions.csv";
+
+	@Override
+	public void submit(final JobCompletion callback, final BlackboardJob bbJob, final String jobName, final JobProfile jobProfile)
+			throws HadoopServiceException {
+
+		final ClusterName clusterName = ClusterName.valueOf(bbJob.getParameters().get("cluster"));
+
+		try {
+			final Properties jobConf = prepareJob(getConf(clusterName), jobName, jobProfile, bbJob.getParameters());
+			log.debug("oozie job configuration:\n" + jobConf);
+
+			if (!hadoopClientMap.isOozieAvailable(clusterName)) throw new HadoopServiceException("oozie not available for cluster: " + clusterName.toString());
+
+			logJobDetails(jobConf);
+
+			final OozieClient oozieClient = hadoopClientMap.getOozieClient(clusterName);
+			final String internalId = oozieClient.run(jobConf);
+			final String jobId = newJobId(clusterName, internalId);
+
+			jobRegistry.registerJob(HadoopJob.newInstance(jobId, clusterName, jobProfile,
+					new OozieJobMonitor(oozieClient, internalId, callback, getWorkflowActions(bbJob))));
+
+		} catch (final OozieClientException e) {
+			throw new HadoopServiceException("error executing hadoop job: " + jobName, e);
+		}
+	}
+
+	/**
+	 * Extracts from the blackboard job params the oozie action names as csv and returns them as a set.
+	 * @param bbJob
+	 * @return
+	 */
+	private Set<String> getWorkflowActions(final BlackboardJob bbJob) {
+		final String actions = bbJob.getParameters().get(OOZIE_REPORT_ACTIONS);
+		if (StringUtils.isNotBlank(actions)) {
+			return Sets.newHashSet(Splitter.on(",").split(actions));
+		}
+		return Sets.newHashSet();
+	}
+
+	private Properties prepareJob(final Configuration configuration, final String jobName, final JobProfile jobProfile, final Map<String, String> parameters) {
+
+		log.info("creating job: " + jobName);
+
+		final Properties p = new Properties();
+
+		merge(p, configuration);
+		merge(p, jobProfile.getJobDefinition().entrySet());
+		merge(p, parameters.entrySet());
+
+		return p;
+	}
+
+	private void merge(final Properties p, final Iterable<Entry<String, String>> entrySet) {
+		for (final Entry<String, String> e : entrySet) {
+			p.setProperty(e.getKey(), e.getValue());
+		}
+	}
+
+	protected void logJobDetails(final Properties jobConf) {
+		for (final Entry<?, ?> e : jobConf.entrySet()) {
+			if (log.isDebugEnabled()) {
+				log.debug("\n" + e.getKey().toString() + " : " + e.getValue().toString());
+			}
+		}
+	}
+
+}
Index: modules/dnet-hadoop-service/tags/dnet-hadoop-service-2.7.8/src/main/java/eu/dnetlib/data/hadoop/action/JobCompletion.java
===================================================================
--- modules/dnet-hadoop-service/tags/dnet-hadoop-service-2.7.8/src/main/java/eu/dnetlib/data/hadoop/action/JobCompletion.java	(nonexistent)
+++ modules/dnet-hadoop-service/tags/dnet-hadoop-service-2.7.8/src/main/java/eu/dnetlib/data/hadoop/action/JobCompletion.java	(revision 59871)
@@ -0,0 +1,16 @@
+package eu.dnetlib.data.hadoop.action;
+
+import java.util.Map;
+
+/**
+ * Callback used to handle hadoop jobs completion.
+ * 
+ * @author claudio
+ * 
+ */
+public interface JobCompletion {
+
+	void done(Map<String, String> properties);
+
+	void failed(String jobId, Throwable e);
+}
Index: modules/dnet-hadoop-service/tags/dnet-hadoop-service-2.7.8/src/main/java/eu/dnetlib/data/hadoop/action/DeleteHdfsPathAction.java
===================================================================
--- modules/dnet-hadoop-service/tags/dnet-hadoop-service-2.7.8/src/main/java/eu/dnetlib/data/hadoop/action/DeleteHdfsPathAction.java	(nonexistent)
+++ modules/dnet-hadoop-service/tags/dnet-hadoop-service-2.7.8/src/main/java/eu/dnetlib/data/hadoop/action/DeleteHdfsPathAction.java	(revision 59871)
@@ -0,0 +1,44 @@
+package eu.dnetlib.data.hadoop.action;
+
+import eu.dnetlib.data.hadoop.HadoopServiceCore;
+import eu.dnetlib.data.hadoop.config.ClusterName;
+import eu.dnetlib.data.hadoop.rmi.HadoopServiceException;
+import eu.dnetlib.enabling.tools.blackboard.BlackboardJob;
+import eu.dnetlib.enabling.tools.blackboard.BlackboardServerHandler;
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.springframework.beans.factory.annotation.Autowired;
+
+/**
+ * Delete a path on hdfs.
+ *
+ * Params - cluster: cluster name, path: the path to deleteFromHBase recursively
+ *
+ * @author claudio
+ *
+ */
+public class DeleteHdfsPathAction extends AbstractHadoopAction {
+
+	/**
+	 * logger.
+	 */
+	private static final Log log = LogFactory.getLog(DeleteHdfsPathAction.class); // NOPMD by marko on 11/24/08 5:02 PM
+
+	/**
+	 * The hadoop service core.
+	 */
+	@Autowired
+	private HadoopServiceCore hadoopServiceCore;
+
+	@Override
+	protected void executeAsync(final BlackboardServerHandler handler, final BlackboardJob job) throws HadoopServiceException {
+		final ClusterName clusterName = ClusterName.valueOf(job.getParameters().get("cluster"));
+		log.info("deleteFromHBase request on cluster: " + clusterName.toString());
+
+		final String path = job.getParameters().get("path");
+		hadoopServiceCore.deleteFromHdfs(clusterName, path);
+
+		handler.done(job);
+	}
+
+}
Index: modules/dnet-hadoop-service/tags/dnet-hadoop-service-2.7.8/src/main/java/eu/dnetlib/data/hadoop/action/SubmitAdminJobAction.java
===================================================================
--- modules/dnet-hadoop-service/tags/dnet-hadoop-service-2.7.8/src/main/java/eu/dnetlib/data/hadoop/action/SubmitAdminJobAction.java	(nonexistent)
+++ modules/dnet-hadoop-service/tags/dnet-hadoop-service-2.7.8/src/main/java/eu/dnetlib/data/hadoop/action/SubmitAdminJobAction.java	(revision 59871)
@@ -0,0 +1,173 @@
+package eu.dnetlib.data.hadoop.action;
+
+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.config.ClusterName;
+import eu.dnetlib.data.hadoop.mapred.MapreduceJobMonitor;
+import eu.dnetlib.data.hadoop.rmi.HadoopJobType.AdminJobType;
+import eu.dnetlib.data.hadoop.rmi.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.lang.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.action.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;
+    }
+
+}
Index: modules/dnet-hadoop-service/tags/dnet-hadoop-service-2.7.8/src/main/java/eu/dnetlib/data/hadoop/action/CreateHdfsDirectoryAction.java
===================================================================
--- modules/dnet-hadoop-service/tags/dnet-hadoop-service-2.7.8/src/main/java/eu/dnetlib/data/hadoop/action/CreateHdfsDirectoryAction.java	(nonexistent)
+++ modules/dnet-hadoop-service/tags/dnet-hadoop-service-2.7.8/src/main/java/eu/dnetlib/data/hadoop/action/CreateHdfsDirectoryAction.java	(revision 59871)
@@ -0,0 +1,54 @@
+package eu.dnetlib.data.hadoop.action;
+
+import eu.dnetlib.data.hadoop.HadoopServiceCore;
+import eu.dnetlib.data.hadoop.config.ClusterName;
+import eu.dnetlib.data.hadoop.rmi.HadoopServiceException;
+import eu.dnetlib.enabling.tools.blackboard.BlackboardJob;
+import eu.dnetlib.enabling.tools.blackboard.BlackboardServerHandler;
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.springframework.beans.factory.annotation.Autowired;
+
+/**
+ * Creates a directory on hdfs, if it does not exists.
+ * <p/>
+ * Params - cluster: cluster name, path: directory path to be created, forceDelete: forces the directory to be dropped and recreated.
+ *
+ * @author claudio
+ */
+public class CreateHdfsDirectoryAction extends AbstractHadoopAction {
+
+	/**
+	 * logger.
+	 */
+	private static final Log log = LogFactory.getLog(CreateHdfsDirectoryAction.class); // NOPMD by marko on 11/24/08 5:02 PM
+
+	/**
+	 * The hadoop service core.
+	 */
+	@Autowired
+	private HadoopServiceCore hadoopServiceCore;
+
+	@Override
+	protected void executeAsync(final BlackboardServerHandler handler, final BlackboardJob job) throws HadoopServiceException {
+
+		final ClusterName clusterName = ClusterName.valueOf(job.getParameters().get("cluster"));
+		log.info("create directory request on cluster: " + clusterName.toString());
+
+		final boolean forceDelete = isForceDelete(job);
+		final String path = job.getParameters().get("path");
+
+		hadoopServiceCore.createHdfsDir(clusterName, path, forceDelete);
+
+		handler.done(job);
+	}
+
+	private boolean isForceDelete(final BlackboardJob job) {
+		try {
+			return Boolean.parseBoolean(job.getParameters().get("force"));
+		} catch (IllegalArgumentException e) {
+			return false;
+		}
+	}
+
+}
Index: modules/dnet-hadoop-service/tags/dnet-hadoop-service-2.7.8/src/main/java/eu/dnetlib/data/hadoop/action/ImportEprHbaseAction.java
===================================================================
--- modules/dnet-hadoop-service/tags/dnet-hadoop-service-2.7.8/src/main/java/eu/dnetlib/data/hadoop/action/ImportEprHbaseAction.java	(nonexistent)
+++ modules/dnet-hadoop-service/tags/dnet-hadoop-service-2.7.8/src/main/java/eu/dnetlib/data/hadoop/action/ImportEprHbaseAction.java	(revision 59871)
@@ -0,0 +1,55 @@
+package eu.dnetlib.data.hadoop.action;
+
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.springframework.beans.factory.annotation.Required;
+
+import eu.dnetlib.data.hadoop.config.ClusterName;
+import eu.dnetlib.data.hadoop.hbase.HbaseTableFeeder;
+import eu.dnetlib.data.hadoop.rmi.HadoopServiceException;
+import eu.dnetlib.enabling.tools.blackboard.BlackboardJob;
+import eu.dnetlib.enabling.tools.blackboard.BlackboardServerHandler;
+import eu.dnetlib.miscutils.functional.xml.DnetXsltFunctions;
+
+public class ImportEprHbaseAction extends AbstractHadoopAction {
+
+	private static final Log log = LogFactory.getLog(ImportEprHbaseAction.class); // NOPMD by marko on 11/24/08 5:02 PM
+
+	private HbaseTableFeeder hbaseTableFeeder;
+
+	@Override
+	public void executeAsync(final BlackboardServerHandler handler, final BlackboardJob job) throws HadoopServiceException {
+
+		final String epr = DnetXsltFunctions.decodeBase64(job.getParameters().get("input_epr"));
+		final String xsl = DnetXsltFunctions.decodeBase64(job.getParameters().get("xslt"));
+		final ClusterName clusterName = ClusterName.valueOf(job.getParameters().get("cluster"));
+		final String table = job.getParameters().get("table");
+		final boolean simulation = Boolean.valueOf(job.getParameters().get("simulation"));
+
+		log.info("Starting import in table " + table);
+
+		try {
+			final Integer count = getHbaseTableFeeder().feed(epr, xsl, clusterName, table, simulation);
+
+			if (count != null) {
+				log.info("Import completed successfully");
+				job.getParameters().put("count", String.valueOf(count));
+			}
+
+			handler.done(job);
+		} catch (final Throwable e) {
+			log.error("Import failed", e);
+			throw new HadoopServiceException("Import failed", e);
+		}
+	}
+
+	public HbaseTableFeeder getHbaseTableFeeder() {
+		return hbaseTableFeeder;
+	}
+
+	@Required
+	public void setHbaseTableFeeder(final HbaseTableFeeder hbaseTableFeeder) {
+		this.hbaseTableFeeder = hbaseTableFeeder;
+	}
+
+}
Index: modules/dnet-hadoop-service/tags/dnet-hadoop-service-2.7.8/src/main/java/eu/dnetlib/data/hadoop/action/SequenceFileFeeder.java
===================================================================
--- modules/dnet-hadoop-service/tags/dnet-hadoop-service-2.7.8/src/main/java/eu/dnetlib/data/hadoop/action/SequenceFileFeeder.java	(nonexistent)
+++ modules/dnet-hadoop-service/tags/dnet-hadoop-service-2.7.8/src/main/java/eu/dnetlib/data/hadoop/action/SequenceFileFeeder.java	(revision 59871)
@@ -0,0 +1,77 @@
+package eu.dnetlib.data.hadoop.action;
+
+import java.io.IOException;
+
+import org.apache.commons.lang.StringUtils;
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.io.SequenceFile;
+import org.apache.hadoop.io.Text;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.beans.factory.annotation.Required;
+
+import eu.dnetlib.data.hadoop.config.ClusterName;
+import eu.dnetlib.data.hadoop.config.ConfigurationEnumerator;
+import eu.dnetlib.data.hadoop.hdfs.SequenceFileWriterFactory;
+import eu.dnetlib.enabling.resultset.client.ResultSetClientFactory;
+
+public class SequenceFileFeeder {
+
+	private static final Log log = LogFactory.getLog(SequenceFileFeeder.class); // NOPMD by marko on 11/24/08 5:02 PM
+
+	private ResultSetClientFactory resultSetClientFactory;
+
+	@Autowired
+	protected ConfigurationEnumerator configurationEnumerator;
+
+	@Autowired
+	protected SequenceFileWriterFactory sequenceFileWriterFactory;
+
+	public int feed(final String epr, final ClusterName clusterName, final String path) throws IOException {
+		return doWrite(epr, clusterName, path);
+	}
+
+	private int doWrite(final String epr, final ClusterName clusterName, final String path) throws IOException {
+		final SequenceFile.Writer writer = sequenceFileWriterFactory.getSequenceFileWriter(Text.class, Text.class, getConf(clusterName), new Path(path));
+		log.info("Opened sequence file writer: " + writer.toString());
+
+		try {
+			final Text idText = new Text();
+			final Text bodyText = new Text();
+			int count = 0;
+			int nulls = 0;
+			for (String record : getResultSetClientFactory().getClient(epr)) {
+				if (StringUtils.isBlank(record)) {
+					nulls++;
+				} else {
+					idText.set(String.valueOf(count++));
+					bodyText.set(record);
+					writer.append(idText, bodyText);
+				}
+			}
+			log.info("written " + count + " records in sequence file: " + path);
+			if (nulls > 0) {
+				log.warn("found " + nulls + " records in epr!");
+			}
+			return count;
+		} finally {
+			writer.close();
+		}
+	}
+
+	protected Configuration getConf(final ClusterName clusterName) {
+		return configurationEnumerator.get(clusterName);
+	}
+
+	public ResultSetClientFactory getResultSetClientFactory() {
+		return resultSetClientFactory;
+	}
+
+	@Required
+	public void setResultSetClientFactory(final ResultSetClientFactory resultSetClientFactory) {
+		this.resultSetClientFactory = resultSetClientFactory;
+	}
+
+}
Index: modules/dnet-hadoop-service/tags/dnet-hadoop-service-2.7.8/src/main/java/eu/dnetlib/data/hadoop/action/JobMonitor.java
===================================================================
--- modules/dnet-hadoop-service/tags/dnet-hadoop-service-2.7.8/src/main/java/eu/dnetlib/data/hadoop/action/JobMonitor.java	(nonexistent)
+++ modules/dnet-hadoop-service/tags/dnet-hadoop-service-2.7.8/src/main/java/eu/dnetlib/data/hadoop/action/JobMonitor.java	(revision 59871)
@@ -0,0 +1,42 @@
+package eu.dnetlib.data.hadoop.action;
+
+import java.util.Date;
+
+import eu.dnetlib.data.hadoop.rmi.HadoopServiceException;
+
+public abstract class JobMonitor implements Runnable {
+
+	private final JobCompletion callback;
+
+	protected Date lastActivity;
+
+    /**
+     * Monitor sleep time in seconds.
+     */
+    protected final int monitorSleepTimeSecs = 10;
+
+	protected JobMonitor(JobCompletion callback) {
+		this.callback = callback;
+		this.lastActivity = new Date(System.currentTimeMillis());
+	}
+
+	@Override
+	public abstract void run();
+
+	public abstract void kill();
+
+	public abstract String getStatus();
+
+	public abstract Date getLastActivity();
+
+	public abstract Date getStartTime() throws HadoopServiceException;
+
+	public abstract String getHadoopId();
+
+	public abstract String getTrackerUrl();
+
+	protected JobCompletion getCallback() {
+		return callback;
+	}
+
+}
Index: modules/dnet-hadoop-service/tags/dnet-hadoop-service-2.7.8/src/main/java/eu/dnetlib/data/hadoop/AbstractHadoopClient.java
===================================================================
--- modules/dnet-hadoop-service/tags/dnet-hadoop-service-2.7.8/src/main/java/eu/dnetlib/data/hadoop/AbstractHadoopClient.java	(nonexistent)
+++ modules/dnet-hadoop-service/tags/dnet-hadoop-service-2.7.8/src/main/java/eu/dnetlib/data/hadoop/AbstractHadoopClient.java	(revision 59871)
@@ -0,0 +1,37 @@
+package eu.dnetlib.data.hadoop;
+
+import eu.dnetlib.data.hadoop.config.ConfigurationEnumerator;
+import eu.dnetlib.data.hadoop.rmi.HadoopServiceException;
+import eu.dnetlib.enabling.is.lookup.rmi.ISLookUpException;
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.springframework.beans.factory.annotation.Autowired;
+
+public abstract class AbstractHadoopClient {
+
+	private static final Log log = LogFactory.getLog(AbstractHadoopClient.class);
+
+	@Autowired
+	private ISClient isClient;
+
+	@Autowired
+	protected ConfigurationEnumerator configurationEnumerator;
+
+	protected void setHadoopUser() throws HadoopServiceException {
+		setHadoopUser(getDefaultUser());
+	}
+
+	private String getDefaultUser() throws HadoopServiceException {
+		try {
+			return isClient.queryForServiceProperty("default.hadoop.user");
+		} catch (ISLookUpException e) {
+			throw new HadoopServiceException(e);
+		}
+	}
+
+	protected void setHadoopUser(final String userName) {
+		log.info("setting HADOOP_USER_NAME as " + userName);
+		System.setProperty("HADOOP_USER_NAME", userName);
+	}
+
+}
Index: modules/dnet-hadoop-service/tags/dnet-hadoop-service-2.7.8/src/main/java/eu/dnetlib/data/hadoop/utils/ScanProperties.java
===================================================================
--- modules/dnet-hadoop-service/tags/dnet-hadoop-service-2.7.8/src/main/java/eu/dnetlib/data/hadoop/utils/ScanProperties.java	(nonexistent)
+++ modules/dnet-hadoop-service/tags/dnet-hadoop-service-2.7.8/src/main/java/eu/dnetlib/data/hadoop/utils/ScanProperties.java	(revision 59871)
@@ -0,0 +1,49 @@
+package eu.dnetlib.data.hadoop.utils;
+
+import java.util.Set;
+
+import org.apache.hadoop.hbase.filter.FilterList;
+import org.apache.hadoop.hbase.filter.FilterList.Operator;
+
+import com.google.common.collect.Sets;
+
+public class ScanProperties {
+
+	private static final Operator DEFAULT_OPERATOR = Operator.MUST_PASS_ALL;
+
+	private int caching = 100;
+	private FilterList filterList;
+	private Set<String> families = Sets.newHashSet();
+
+	public ScanProperties(final String op) {
+		Operator operator = DEFAULT_OPERATOR;
+		if ((op != null) && !op.isEmpty()) {
+			operator = Operator.valueOf(op);
+		}
+		filterList = new FilterList(operator);
+	}
+
+	public FilterList getFilterList() {
+		return filterList;
+	}
+
+	public void setFilterList(final FilterList filterList) {
+		this.filterList = filterList;
+	}
+
+	public Set<String> getFamilies() {
+		return families;
+	}
+
+	public void setFamilies(final Set<String> families) {
+		this.families = families;
+	}
+
+	public int getCaching() {
+		return caching;
+	}
+
+	public void setCaching(int caching) {
+		this.caching = caching;
+	}
+}
Index: modules/dnet-hadoop-service/tags/dnet-hadoop-service-2.7.8/src/main/java/eu/dnetlib/data/hadoop/utils/JobProfile.java
===================================================================
--- modules/dnet-hadoop-service/tags/dnet-hadoop-service-2.7.8/src/main/java/eu/dnetlib/data/hadoop/utils/JobProfile.java	(nonexistent)
+++ modules/dnet-hadoop-service/tags/dnet-hadoop-service-2.7.8/src/main/java/eu/dnetlib/data/hadoop/utils/JobProfile.java	(revision 59871)
@@ -0,0 +1,67 @@
+package eu.dnetlib.data.hadoop.utils;
+
+import java.util.Map;
+import java.util.Set;
+
+import com.google.common.collect.Maps;
+import com.google.common.collect.Sets;
+
+import eu.dnetlib.data.hadoop.rmi.HadoopJobType;
+
+public class JobProfile {
+
+	private final Map<String, String> jobDefinition = Maps.newHashMap();
+	private final Set<String> requiredParams = Sets.newHashSet();
+	private ScanProperties scanProperties;
+
+	private String name;
+
+	private String description = "";
+
+	private HadoopJobType jobType;
+
+	public boolean isEmpty() {
+		return getJobDefinition().isEmpty();
+	}
+
+	public Map<String, String> getJobDefinition() {
+		return jobDefinition;
+	}
+
+	public Set<String> getRequiredParams() {
+		return requiredParams;
+	}
+
+	public String getName() {
+		return name;
+	}
+
+	public void setName(String name) {
+		this.name = name;
+	}
+
+	public String getDescription() {
+		return description;
+	}
+
+	public void setDescription(String description) {
+		this.description = description;
+	}
+
+	public ScanProperties getScanProperties() {
+		return scanProperties;
+	}
+
+	public void setScanProperties(ScanProperties scanProperties) {
+		this.scanProperties = scanProperties;
+	}
+
+	public HadoopJobType getJobType() {
+		return jobType;
+	}
+
+	public void setJobType(HadoopJobType jobType) {
+		this.jobType = jobType;
+	}
+
+}
Index: modules/dnet-hadoop-service/tags/dnet-hadoop-service-2.7.8/src/test/java/eu/dnetlib/data/hadoop/hbase/HBaseTestContextConfiguration.java
===================================================================
--- modules/dnet-hadoop-service/tags/dnet-hadoop-service-2.7.8/src/test/java/eu/dnetlib/data/hadoop/hbase/HBaseTestContextConfiguration.java	(nonexistent)
+++ modules/dnet-hadoop-service/tags/dnet-hadoop-service-2.7.8/src/test/java/eu/dnetlib/data/hadoop/hbase/HBaseTestContextConfiguration.java	(revision 59871)
@@ -0,0 +1,80 @@
+package eu.dnetlib.data.hadoop.hbase;
+
+import eu.dnetlib.data.hadoop.HadoopClientMap;
+import eu.dnetlib.data.hadoop.HadoopServiceCore;
+import eu.dnetlib.data.hadoop.config.ConfigurationEnumerator;
+import eu.dnetlib.data.hadoop.config.ConfigurationFactory;
+import eu.dnetlib.data.hadoop.mapred.JobClientFactory;
+import eu.dnetlib.data.hadoop.oozie.OozieClientFactory;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.context.annotation.Profile;
+import org.springframework.core.io.ClassPathResource;
+import org.springframework.core.io.Resource;
+
+@Configuration
+@Profile(value = "test")
+public class HBaseTestContextConfiguration {
+
+	public static final String ENABLED_CLIENTS = "{"
+			+ "\"DM\":{\"oozie\":\"false\",\"mapred\":\"false\",\"hbase\":\"true\"},"
+			+ "\"IIS\":{\"oozie\":\"false\",\"mapred\":\"false\",\"hbase\":\"false\"}"
+			+ "}";
+
+	public static final int MAX_VERSIONS = 10;
+
+	@Bean
+	public HadoopServiceCore hadoopServiceCore() {
+		final HadoopServiceCore core = new HadoopServiceCore();
+
+		core.setMaxVersions(MAX_VERSIONS);
+
+		System.out.println("using hbase max versions: " + MAX_VERSIONS);
+		return core;
+	}
+
+	@Bean
+	public HadoopClientMap hadoopClientMap() throws InterruptedException {
+		final HadoopClientMap clientMap = new HadoopClientMap();
+		clientMap.setEnabledClients(ENABLED_CLIENTS);
+
+		return clientMap;
+	}
+
+	@Bean
+	public HBaseAdminFactory hBaseAdminFactory() {
+		return new HBaseAdminFactory();
+	}
+
+	@Bean
+	public OozieClientFactory oozieClientFactory() {
+		return new OozieClientFactory();
+	}
+
+	@Bean
+	public JobClientFactory jobClientFactory() {
+		return new JobClientFactory();
+	}
+
+	@Bean
+	public ConfigurationEnumerator configurationEnumerator() {
+		return new ConfigurationEnumerator();
+	}
+
+	@Bean
+	public ConfigurationFactory DM() {
+		return get(new ClassPathResource("/eu/dnetlib/data/hadoop/config/hadoop-default.dm.cnr.properties"));
+	}
+
+	@Bean
+	public ConfigurationFactory IIS() {
+		return get(new ClassPathResource("/eu/dnetlib/data/hadoop/config/hadoop-default.iis.icm.properties"));
+	}
+
+	protected ConfigurationFactory get(final Resource props) {
+		final ConfigurationFactory configurationFactory = new ConfigurationFactory();
+		configurationFactory.setDefaults(props);
+		return configurationFactory;
+	}
+
+}
Index: modules/dnet-hadoop-service/tags/dnet-hadoop-service-2.7.8/src/test/java/eu/dnetlib/data/hadoop/utils/CopyTableTest.java
===================================================================
--- modules/dnet-hadoop-service/tags/dnet-hadoop-service-2.7.8/src/test/java/eu/dnetlib/data/hadoop/utils/CopyTableTest.java	(nonexistent)
+++ modules/dnet-hadoop-service/tags/dnet-hadoop-service-2.7.8/src/test/java/eu/dnetlib/data/hadoop/utils/CopyTableTest.java	(revision 59871)
@@ -0,0 +1,30 @@
+package eu.dnetlib.data.hadoop.utils;
+
+import static org.junit.Assert.assertNotNull;
+
+import java.io.IOException;
+import java.util.Map.Entry;
+
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.hbase.mapreduce.CopyTable;
+import org.apache.hadoop.mapreduce.Job;
+import org.junit.Test;
+
+public class CopyTableTest {
+
+	@Test
+	public void testCopyTable() throws IOException {
+
+		Job job =
+				CopyTable.createSubmittableJob(new Configuration(), new String[] { "--peer.adr=server1,server2,server3:2181:/hbase",
+					"--families=myOldCf:myNewCf,cf2,cf3", "tableName" });
+		assertNotNull(job);
+
+		Configuration conf = job.getConfiguration();
+
+		for (Entry<String, String> e : conf) {
+			System.out.println(String.format("<PROPERTY key=\"%s\" value=\"%s\"/>", e.getKey(), e.getValue()));
+		}
+
+	}
+}
Index: modules/dnet-hadoop-service/tags/dnet-hadoop-service-2.7.8
===================================================================
--- modules/dnet-hadoop-service/tags/dnet-hadoop-service-2.7.8	(nonexistent)
+++ modules/dnet-hadoop-service/tags/dnet-hadoop-service-2.7.8	(revision 59871)

Property changes on: modules/dnet-hadoop-service/tags/dnet-hadoop-service-2.7.8
___________________________________________________________________
Added: svn:ignore
## -0,0 +1,7 ##
+target
+.settings
+.classpath
+.project
+
+
+*.iml
