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 60326)
+++ modules/dnet-openaireplus-workflows/trunk/src/main/java/eu/dnetlib/msro/openaireplus/workflows/nodes/dhp/PrepareEnvCollectHadoopJobNode.java	(revision 60327)
@@ -1,11 +1,18 @@
 package eu.dnetlib.msro.openaireplus.workflows.nodes.dhp;
+import java.text.SimpleDateFormat;
+import java.util.Date;
+import java.util.Iterator;
 import java.util.Map;
 import java.util.Optional;
 import java.util.stream.Collectors;
 
+import eu.dnetlib.common.logging.DnetLogger;
 import eu.dnetlib.dhp.model.mdstore.Provenance;
 import eu.dnetlib.enabling.datasources.common.Datasource;
 import eu.dnetlib.enabling.datasources.common.DsmNotFoundException;
+import eu.dnetlib.msro.rmi.MSROException;
+import org.apache.commons.lang.math.NumberUtils;
+import org.apache.commons.lang3.StringUtils;
 import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
 import org.springframework.beans.factory.annotation.Autowired;
@@ -21,11 +28,20 @@
 import eu.dnetlib.msro.workflows.util.WorkflowsConstants;
 import org.springframework.beans.factory.annotation.Value;
 
+import javax.annotation.Resource;
+
 public class PrepareEnvCollectHadoopJobNode extends SimpleJobNode {
 
     private static final Log log = LogFactory.getLog(PrepareEnvCollectHadoopJobNode.class);
+
     public static final String METADATA_IDENTIFIER_PATH = "metadata_identifier_path";
+    public static final String DATE_FORMAT = "yyyy-MM-dd";
 
+    private long ONE_DAY = 1000 * 60 * 60 * 24;
+
+    @Resource(name = "msroWorkflowLogger")
+    private DnetLogger dnetLogger;
+
     @Autowired
     private LocalDatasourceManager<?, ?> dsManager;
 
@@ -40,6 +56,11 @@
     private String collectionMode;
 
     /**
+     * used in INCREMENTAL mode
+     */
+    private String fromDateOverride;
+
+    /**
      * XML | JSON, default = XML
      */
     private String metadataEncoding = "XML";
@@ -50,10 +71,6 @@
     @Override
     protected String execute(final NodeToken token) throws Exception {
 
-        // param 1 : hdfs path
-        // param 2 : api descriptor (json)
-        // param 3 : nameNode
-
         final String dsId = token.getEnv().getAttribute(WorkflowsConstants.DATAPROVIDER_ID);
         log.info("dsId: " + dsId);
         final String apiId = token.getEnv().getAttribute(WorkflowsConstants.DATAPROVIDER_INTERFACE);
@@ -83,6 +100,16 @@
             token.getEnv().setAttribute("oozieWfPath", getOozieWfPath());
 
             final ApiDescriptor api = opt.get();
+            if ("INCREMENTAL".equals(getCollectionMode())) {
+                final String fromDate = calculateFromDate(token);
+
+                log.info("Incremental Harvesting from: " + fromDate);
+
+                if (StringUtils.isNotBlank(fromDate)) {
+                   api.getParams().put("fromDate", fromDate);
+                }
+            }
+
             token.getEnv().setAttribute("apiDescription", new Gson().toJson(api));
 
             final Provenance provenance = new Provenance();
@@ -113,6 +140,51 @@
         }
     }
 
+    private String findCurrentWfProfileId(NodeToken token) throws MSROException {
+        final String p1 = token.getEnv().getAttribute(WorkflowsConstants.SYSTEM_WF_PROFILE_ID);
+        if (p1 != null && !p1.isEmpty()) {
+            return p1;
+        }
+        final String p2 = token.getFullEnv().getAttribute(WorkflowsConstants.SYSTEM_WF_PROFILE_ID);
+        if (p2 != null && !p2.isEmpty()) {
+            return p2;
+        }
+        final String p3 = token.getProcess().getEnv().getAttribute(WorkflowsConstants.SYSTEM_WF_PROFILE_ID);
+        if (p3 != null && !p3.isEmpty()) {
+            return p3;
+        }
+        throw new MSROException("Missing property in env: " + WorkflowsConstants.SYSTEM_WF_PROFILE_ID);
+    }
+
+    private String calculateFromDate(final NodeToken token) throws MSROException {
+
+        if (StringUtils.isNotBlank(getFromDateOverride())) {
+            log.info("using override FROM_DATE for incremental harvesting: " + getFromDateOverride());
+            return getFromDateOverride();
+        }
+        final String profId = findCurrentWfProfileId(token);
+
+        final long d = findLastSuccessStartDate(profId);
+        return (d > 0) ? (new SimpleDateFormat(DATE_FORMAT)).format(new Date(d - ONE_DAY)) : null;
+    }
+
+    private long findLastSuccessStartDate(String profId) {
+        long res = -1;
+
+        final Iterator<Map<String, String>> iter = dnetLogger.find(WorkflowsConstants.SYSTEM_WF_PROFILE_ID, profId);
+        while (iter.hasNext()) {
+            final Map<String, String> map = iter.next();
+            if ("true".equalsIgnoreCase(map.get(WorkflowsConstants.SYSTEM_COMPLETED_SUCCESSFULLY))) {
+                final long curr = NumberUtils.toLong(map.get(WorkflowsConstants.SYSTEM_START_DATE), -1);
+                if (curr > res) {
+                    res = curr;
+                }
+            }
+        }
+        return res;
+    }
+
+
     public String getMdId() {
         return mdId;
     }
@@ -144,4 +216,12 @@
     public void setMetadataEncoding(String metadataEncoding) {
         this.metadataEncoding = metadataEncoding;
     }
+
+    public String getFromDateOverride() {
+        return fromDateOverride;
+    }
+
+    public void setFromDateOverride(String fromDateOverride) {
+        this.fromDateOverride = fromDateOverride;
+    }
 }
Index: modules/dnet-openaireplus-workflows/trunk/src/main/resources/eu/dnetlib/msro/openaireplus/workflows/repo-hi/pubsRepositoryHadoop/collection.wf.st
===================================================================
--- modules/dnet-openaireplus-workflows/trunk/src/main/resources/eu/dnetlib/msro/openaireplus/workflows/repo-hi/pubsRepositoryHadoop/collection.wf.st	(revision 60326)
+++ modules/dnet-openaireplus-workflows/trunk/src/main/resources/eu/dnetlib/msro/openaireplus/workflows/repo-hi/pubsRepositoryHadoop/collection.wf.st	(revision 60327)
@@ -26,6 +26,7 @@
         <PARAM required="true" type="string" name="mdId" managedBy="system" category="MDSTORE_ID">$params.("harv_id")$</PARAM>
         <PARAM required="true" type="string" name="collectionMode" managedBy="user" function="validValues(['REFRESH','INCREMENTAL'])"></PARAM>
         <PARAM required="true" type="string" name="metadataEncoding" managedBy="user">XML</PARAM>
+        <PARAM managedBy="user" name="fromDateOverride" required="false" type="string"/>
     </PARAMETERS>
     <ARCS>
         <ARC to="COLLECT_HADOOP"/>
