Index: modules/dnet-openaireplus-workflows/trunk/src/main/java/eu/dnetlib/msro/openaireplus/workflows/nodes/dhp/PrepareEnvCollectHadoopJobNode.java
===================================================================
--- modules/dnet-openaireplus-workflows/trunk/src/main/java/eu/dnetlib/msro/openaireplus/workflows/nodes/dhp/PrepareEnvCollectHadoopJobNode.java	(revision 60987)
+++ modules/dnet-openaireplus-workflows/trunk/src/main/java/eu/dnetlib/msro/openaireplus/workflows/nodes/dhp/PrepareEnvCollectHadoopJobNode.java	(revision 60988)
@@ -107,7 +107,7 @@
     @Override
     protected String execute(final NodeToken token) throws Exception {
 
-        final String dsId = token.getEnv().getAttribute(WorkflowsConstants.DATAPROVIDER_ID);
+        final String dsId = token.getEnv().getAttribute("parentDatasourceId");
         log.info("dsId: " + dsId);
         final String apiId = token.getEnv().getAttribute(WorkflowsConstants.DATAPROVIDER_INTERFACE);
         log.info("apiId: " + apiId);
Index: modules/dnet-openaireplus-workflows/trunk/src/main/java/eu/dnetlib/msro/openaireplus/workflows/nodes/dhp/CreateMDStoreHadoopJobNode.java
===================================================================
--- modules/dnet-openaireplus-workflows/trunk/src/main/java/eu/dnetlib/msro/openaireplus/workflows/nodes/dhp/CreateMDStoreHadoopJobNode.java	(revision 60987)
+++ modules/dnet-openaireplus-workflows/trunk/src/main/java/eu/dnetlib/msro/openaireplus/workflows/nodes/dhp/CreateMDStoreHadoopJobNode.java	(revision 60988)
@@ -4,6 +4,10 @@
 import java.util.Map;
 
 import com.google.gson.Gson;
+import eu.dnetlib.enabling.is.lookup.rmi.ISLookUpService;
+import eu.dnetlib.enabling.locators.UniqueServiceLocator;
+import eu.dnetlib.msro.rmi.MSROException;
+import org.apache.commons.lang3.StringUtils;
 import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
 import org.springframework.beans.factory.annotation.Required;
@@ -17,6 +21,8 @@
 import eu.dnetlib.msro.workflows.nodes.SimpleJobNode;
 import eu.dnetlib.msro.workflows.util.WorkflowsConstants;
 
+import javax.annotation.Resource;
+
 public class CreateMDStoreHadoopJobNode extends SimpleJobNode {
 
     private static final Log log = LogFactory.getLog(CreateMDStoreHadoopJobNode.class);
@@ -30,6 +36,9 @@
     /* Spring managed params */
     private String mdStoreManagerUrl;
 
+    @Resource
+    private UniqueServiceLocator serviceLocator;
+
     @Override
     protected String execute(final NodeToken token) throws Exception {
         final String url = getMdStoreManagerUrl() + "/new/{format}/{layout}/{interpretation}";
@@ -39,9 +48,24 @@
         params.put("layout", getLayout());
         params.put("interpretation", getInterpretation());
 
+        final String repositoryId = token.getFullEnv().getAttribute(WorkflowsConstants.DATAPROVIDER_ORIGINALID);
+        final String xquery = String.format(
+                "for $x in collection('/db/DRIVER/RepositoryServiceResources/RepositoryServiceResourceType') \n" +
+                "where .//RESOURCE_IDENTIFIER[./@value='%s']\n" +
+                "return $x//EXTRA_FIELDS/FIELD[./key/text() = 'OpenAireDataSourceId']/value/text()",
+                repositoryId);
+
+        final String openAireDataSourceId = serviceLocator.getService(ISLookUpService.class).getResourceProfileByQuery(xquery);
+
+        if (StringUtils.isBlank(openAireDataSourceId)) {
+            throw new MSROException("unable to find openAireDataSourceId in repository profile " + repositoryId);
+        }
+
+        log.info(String.format("got OpenAIRE datasource id '%s' for repository profile '%s'", openAireDataSourceId, repositoryId));
+
         final URI uri = UriComponentsBuilder.fromUriString(url)
                 .queryParam("dsName", token.getFullEnv().getAttribute(WorkflowsConstants.DATAPROVIDER_NAME))
-                .queryParam("dsId", token.getFullEnv().getAttribute(WorkflowsConstants.DATAPROVIDER_ORIGINALID))
+                .queryParam("dsId", openAireDataSourceId)
                 .queryParam("apiId", token.getFullEnv().getAttribute(WorkflowsConstants.DATAPROVIDER_INTERFACE))
                 .buildAndExpand(params)
                 .toUri();
Index: modules/dnet-openaireplus-workflows/trunk/src/main/resources/eu/dnetlib/test/profiles/openaireplus/workflows/repo-hi/Aggregate_Metadata_from_PubsRepository_Hadoop.xml
===================================================================
--- modules/dnet-openaireplus-workflows/trunk/src/main/resources/eu/dnetlib/test/profiles/openaireplus/workflows/repo-hi/Aggregate_Metadata_from_PubsRepository_Hadoop.xml	(revision 60987)
+++ modules/dnet-openaireplus-workflows/trunk/src/main/resources/eu/dnetlib/test/profiles/openaireplus/workflows/repo-hi/Aggregate_Metadata_from_PubsRepository_Hadoop.xml	(revision 60988)
@@ -10,7 +10,7 @@
         <WORKFLOW_NAME>Aggregate Metadata DublinCore format [Hadoop]</WORKFLOW_NAME>
         <WORKFLOW_INFO>
             <FIELD name="Action">Aggregate Metadata</FIELD>
-            <FIELD name="Consequence IS">Support</FIELD>
+            <FIELD name="Consequence IS">Ingestion</FIELD>
             <FIELD name="Datasource class">PubsRepository</FIELD>
             <FIELD name="Content">publications</FIELD>
         </WORKFLOW_INFO>
Index: modules/dnet-openaireplus-workflows/trunk/src/main/resources/eu/dnetlib/test/profiles/openaireplus/workflows/repo-hi/Aggregate_Metadata_from_DataRepository_Hadoop.xml
===================================================================
--- modules/dnet-openaireplus-workflows/trunk/src/main/resources/eu/dnetlib/test/profiles/openaireplus/workflows/repo-hi/Aggregate_Metadata_from_DataRepository_Hadoop.xml	(revision 60987)
+++ modules/dnet-openaireplus-workflows/trunk/src/main/resources/eu/dnetlib/test/profiles/openaireplus/workflows/repo-hi/Aggregate_Metadata_from_DataRepository_Hadoop.xml	(revision 60988)
@@ -10,7 +10,7 @@
         <WORKFLOW_NAME>Aggregate Metadata Datacite format [Hadoop]</WORKFLOW_NAME>
         <WORKFLOW_INFO>
             <FIELD name="Action">Aggregate Metadata</FIELD>
-            <FIELD name="Consequence IS">Support</FIELD>
+            <FIELD name="Consequence IS">Ingestion</FIELD>
             <FIELD name="Datasource class">PubsRepository</FIELD>
             <FIELD name="Content">publications</FIELD>
         </WORKFLOW_INFO>
