map datasource identities (including piwik ids) as original IDs

dhp-build/dhp-build-assembly-resources/pom.xml

@@ -6,7 +6,7 @@
     <parent>
         <groupId>eu.dnetlib.dhp</groupId>
         <artifactId>dhp-build</artifactId>
-        <version>1.2.2-SNAPSHOT</version>
+        <version>1.2.3-SNAPSHOT</version>
     </parent>
 
     <artifactId>dhp-build-assembly-resources</artifactId>

dhp-build/dhp-build-properties-maven-plugin/pom.xml

@@ -6,7 +6,7 @@
     <parent>
         <groupId>eu.dnetlib.dhp</groupId>
         <artifactId>dhp-build</artifactId>
-        <version>1.2.2-SNAPSHOT</version>
+        <version>1.2.3-SNAPSHOT</version>
     </parent>
 
     <artifactId>dhp-build-properties-maven-plugin</artifactId>

dhp-build/dhp-code-style/pom.xml

@@ -5,7 +5,7 @@
 
     <groupId>eu.dnetlib.dhp</groupId>
     <artifactId>dhp-code-style</artifactId>
-    <version>1.2.2-SNAPSHOT</version>
+    <version>1.2.3-SNAPSHOT</version>
 
     <packaging>jar</packaging>
 

dhp-build/pom.xml

@@ -4,7 +4,7 @@
 	<parent>
 		<groupId>eu.dnetlib.dhp</groupId>
 		<artifactId>dhp</artifactId>
-		<version>1.2.2-SNAPSHOT</version>
+		<version>1.2.3-SNAPSHOT</version>
 	</parent>
 	<artifactId>dhp-build</artifactId>
 	<packaging>pom</packaging>

dhp-common/pom.xml

@@ -5,7 +5,7 @@
 	<parent>
 		<groupId>eu.dnetlib.dhp</groupId>
 		<artifactId>dhp</artifactId>
-		<version>1.2.2-SNAPSHOT</version>
+		<version>1.2.3-SNAPSHOT</version>
 		<relativePath>../</relativePath>
 	</parent>
 

dhp-common/src/main/java/eu/dnetlib/dhp/common/FunctionalInterfaceSupport.java

@@ -2,6 +2,7 @@
 package eu.dnetlib.dhp.common;
 
 import java.io.Serializable;
+import java.util.function.Consumer;
 import java.util.function.Supplier;
 
 /** Provides serializable and throwing extensions to standard functional interfaces. */
@@ -10,6 +11,16 @@ public class FunctionalInterfaceSupport {
 	private FunctionalInterfaceSupport() {
 	}
 
+	/**
+	 * Serializable consumer of any kind of objects. To be used withing spark processing pipelines when supplying
+	 * functions externally.
+	 *
+	 * @param <T>
+	 */
+	@FunctionalInterface
+	public interface SerializableConsumer<T> extends Consumer<T>, Serializable {
+	}
+
 	/**
 	 * Serializable supplier of any kind of objects. To be used withing spark processing pipelines when supplying
 	 * functions externally.

dhp-common/src/main/java/eu/dnetlib/dhp/common/PacePerson.java

@@ -17,12 +17,11 @@ import com.google.common.collect.Lists;
 import com.google.common.hash.Hashing;
 
 /**
-* PacePerson tries to derive information from the fullname string of an author.
-* Such informations are Names, Surnames an Fullname split into terms. It provides also an additional field for
-* the original data.
-* The calculation of the names and the surnames is not always possible. When it is impossible to assert which are the
-* names and the surnames, the lists are empty.
-* */
+ * PacePerson tries to derive information from the fullname string of an author. Such informations are Names, Surnames
+ * an Fullname split into terms. It provides also an additional field for the original data. The calculation of the
+ * names and the surnames is not always possible. When it is impossible to assert which are the names and the surnames,
+ * the lists are empty.
+ */
 public class PacePerson {
 
 	private static final String UTF8 = "UTF-8";
@@ -34,18 +33,18 @@ public class PacePerson {
 	private static Set<String> particles = null;
 
 	/**
-	* Capitalizes a string
+	 * Capitalizes a string
 	 *
 	 * @param s the string to capitalize
 	 * @return the input string with capital letter
-	* */
+	 */
 	public static final String capitalize(final String s) {
 		return WordUtils.capitalize(s.toLowerCase(), ' ', '-');
 	}
 
 	/**
-	* Adds a dot to a string with length equals to 1
-	* */
+	 * Adds a dot to a string with length equals to 1
+	 */
 	public static final String dotAbbreviations(final String s) {
 		return s.length() == 1 ? s + "." : s;
 	}
@@ -63,11 +62,11 @@ public class PacePerson {
 	}
 
 	/**
-	* The constructor of the class. It fills the fields of the class basing on the input fullname.
+	 * The constructor of the class. It fills the fields of the class basing on the input fullname.
 	 *
 	 * @param s the input string (fullname of the author)
 	 * @param aggressive set the string normalization type
-	* */
+	 */
 	public PacePerson(String s, final boolean aggressive) {
 		original = s;
 		s = Normalizer.normalize(s, Normalizer.Form.NFD);
@@ -86,7 +85,7 @@ public class PacePerson {
 			// s = s.replaceAll("[\\W&&[^,-]]", "");
 		}
 
-		//if the string contains a comma, it can derive surname and name by splitting on it
+		// if the string contains a comma, it can derive surname and name by splitting on it
 		if (s.contains(",")) {
 			final String[] arr = s.split(",");
 			if (arr.length == 1) {
@@ -97,23 +96,23 @@ public class PacePerson {
 				fullname.addAll(surname);
 				fullname.addAll(name);
 			}
-		} else {  //otherwise, it should rely on CAPS terms and short terms
+		} else { // otherwise, it should rely on CAPS terms and short terms
 			fullname = splitTerms(s);
 
 			int lastInitialPosition = fullname.size();
 			boolean hasSurnameInUpperCase = false;
 
-			//computes lastInitialPosition and hasSurnameInUpperCase
+			// computes lastInitialPosition and hasSurnameInUpperCase
 			for (int i = 0; i < fullname.size(); i++) {
 				final String term = fullname.get(i);
 				if (term.length() == 1) {
-					lastInitialPosition = i; //first word in the name longer than 1 (to avoid name with dots)
+					lastInitialPosition = i; // first word in the name longer than 1 (to avoid name with dots)
 				} else if (term.equals(term.toUpperCase())) {
-					hasSurnameInUpperCase = true; //if one of the words is CAPS
+					hasSurnameInUpperCase = true; // if one of the words is CAPS
 				}
 			}
 
-			//manages particular cases of fullnames
+			// manages particular cases of fullnames
 			if (lastInitialPosition < fullname.size() - 1) { // Case: Michele G. Artini
 				name = fullname.subList(0, lastInitialPosition + 1);
 				surname = fullname.subList(lastInitialPosition + 1, fullname.size());

dhp-common/src/test/java/eu/dnetlib/dhp/common/PacePersonTest.java

@@ -1,26 +1,27 @@
-package eu.dnetlib.dhp.common;
 
-import org.junit.jupiter.api.Test;
+package eu.dnetlib.dhp.common;
 
 import static org.junit.jupiter.api.Assertions.*;
 
+import org.junit.jupiter.api.Test;
+
 public class PacePersonTest {
 
-    @Test
-    public void pacePersonTest1(){
+	@Test
+	public void pacePersonTest1() {
 
-        PacePerson p = new PacePerson("Artini, Michele", false);
-        assertEquals("Artini",p.getSurnameString());
-        assertEquals("Michele", p.getNameString());
-        assertEquals("Artini, Michele", p.getNormalisedFullname());
-    }
+		PacePerson p = new PacePerson("Artini, Michele", false);
+		assertEquals("Artini", p.getSurnameString());
+		assertEquals("Michele", p.getNameString());
+		assertEquals("Artini, Michele", p.getNormalisedFullname());
+	}
 
-    @Test
-    public void pacePersonTest2(){
-        PacePerson p = new PacePerson("Michele G. Artini", false);
-        assertEquals("Artini, Michele G.", p.getNormalisedFullname());
-        assertEquals("Michele G", p.getNameString());
-        assertEquals("Artini", p.getSurnameString());
-    }
+	@Test
+	public void pacePersonTest2() {
+		PacePerson p = new PacePerson("Michele G. Artini", false);
+		assertEquals("Artini, Michele G.", p.getNormalisedFullname());
+		assertEquals("Michele G", p.getNameString());
+		assertEquals("Artini", p.getSurnameString());
+	}
 
 }

dhp-schemas/pom.xml

@@ -5,7 +5,7 @@
     <parent>
         <groupId>eu.dnetlib.dhp</groupId>
         <artifactId>dhp</artifactId>
-        <version>1.2.2-SNAPSHOT</version>
+        <version>1.2.3-SNAPSHOT</version>
         <relativePath>../</relativePath>
     </parent>
 

dhp-schemas/src/main/java/eu/dnetlib/dhp/schema/common/ModelConstants.java

@@ -14,6 +14,7 @@ public class ModelConstants {
 	public static final String DNET_DATA_CITE_RESOURCE = "dnet:dataCite_resource";
 	public static final String DNET_PROVENANCE_ACTIONS = "dnet:provenanceActions";
 	public static final String DNET_COUNTRY_TYPE = "dnet:countries";
+	public static final String DNET_REVIEW_LEVELS = "dnet:review_levels";
 
 	public static final String SYSIMPORT_CROSSWALK_REPOSITORY = "sysimport:crosswalk:repository";
 	public static final String SYSIMPORT_CROSSWALK_ENTITYREGISTRY = "sysimport:crosswalk:entityregistry";
@@ -25,6 +26,10 @@ public class ModelConstants {
 	public static final String ORP_RESULTTYPE_CLASSID = "other";
 
 	public static final String RESULT_RESULT = "resultResult";
+	/**
+	 * @deprecated Use {@link ModelConstants#RELATIONSHIP} instead.
+	 */
+	@Deprecated
 	public static final String PUBLICATION_DATASET = "publicationDataset";
 	public static final String IS_RELATED_TO = "isRelatedTo";
 	public static final String SUPPLEMENT = "supplement";
@@ -34,6 +39,12 @@ public class ModelConstants {
 	public static final String IS_PART_OF = "IsPartOf";
 	public static final String HAS_PARTS = "HasParts";
 	public static final String RELATIONSHIP = "relationship";
+	public static final String CITATION = "citation";
+	public static final String CITES = "cites";
+	public static final String IS_CITED_BY = "IsCitedBy";
+	public static final String REVIEW = "review";
+	public static final String REVIEWS = "reviews";
+	public static final String IS_REVIEWED_BY = "IsReviewedBy";
 
 	public static final String RESULT_PROJECT = "resultProject";
 	public static final String OUTCOME = "outcome";

dhp-schemas/src/main/java/eu/dnetlib/dhp/schema/oaf/Instance.java

@@ -31,7 +31,7 @@ public class Instance implements Serializable {
 	// typed results
 	private Field<String> processingchargecurrency;
 
-	private Field<String> refereed; // peer-review status
+	private Qualifier refereed; // peer-review status
 
 	public Field<String> getLicense() {
 		return license;
@@ -113,11 +113,11 @@ public class Instance implements Serializable {
 		this.processingchargecurrency = processingchargecurrency;
 	}
 
-	public Field<String> getRefereed() {
+	public Qualifier getRefereed() {
 		return refereed;
 	}
 
-	public void setRefereed(Field<String> refereed) {
+	public void setRefereed(Qualifier refereed) {
 		this.refereed = refereed;
 	}
 

dhp-schemas/src/main/java/eu/dnetlib/dhp/schema/oaf/Result.java

@@ -254,28 +254,25 @@ public class Result extends OafEntity implements Serializable {
 				final StructuredProperty p = baseMainTitle;
 				title = title.stream().filter(t -> t != p).collect(Collectors.toList());
 			}
-//
-//
-//			title.remove(baseMainTitle);
 		}
 
 		StructuredProperty newMainTitle = null;
 		if (r.getTitle() != null) {
 			newMainTitle = getMainTitle(r.getTitle());
-			if (newMainTitle != null) {
+			if (newMainTitle != null && title != null) {
 				final StructuredProperty p = newMainTitle;
 				title = title.stream().filter(t -> t != p).collect(Collectors.toList());
 			}
-
-			// r.getTitle().remove(newMainTitle);
 		}
 
-		if (newMainTitle != null && compareTrust(this, r) < 0)
+		if (newMainTitle != null && compareTrust(this, r) < 0) {
 			baseMainTitle = newMainTitle;
+		}
 
 		title = mergeLists(title, r.getTitle());
-		if (title != null && baseMainTitle != null)
+		if (title != null && baseMainTitle != null) {
 			title.add(baseMainTitle);
+		}
 
 		relevantdate = mergeLists(relevantdate, r.getRelevantdate());
 

dhp-workflows/dhp-actionmanager/pom.xml

@@ -4,7 +4,7 @@
     <parent>
         <groupId>eu.dnetlib.dhp</groupId>
         <artifactId>dhp-workflows</artifactId>
-        <version>1.2.2-SNAPSHOT</version>
+        <version>1.2.3-SNAPSHOT</version>
     </parent>
     <artifactId>dhp-actionmanager</artifactId>
 

dhp-workflows/dhp-actionmanager/src/main/java/eu/dnetlib/dhp/actionmanager/migration/ProtoConverter.java

@@ -96,12 +96,21 @@ public class ProtoConverter implements Serializable {
 					.stream()
 					.distinct()
 					.collect(Collectors.toCollection(ArrayList::new)) : null);
-		i.setRefereed(mapStringField(ri.getRefereed()));
+		i.setRefereed(mapRefereed(ri.getRefereed()));
 		i.setProcessingchargeamount(mapStringField(ri.getProcessingchargeamount()));
 		i.setProcessingchargecurrency(mapStringField(ri.getProcessingchargecurrency()));
 		return i;
 	}
 
+	private static Qualifier mapRefereed(FieldTypeProtos.StringField refereed) {
+		Qualifier q = new Qualifier();
+		q.setClassid(refereed.getValue());
+		q.setSchemename(refereed.getValue());
+		q.setSchemeid("dnet:review_levels");
+		q.setSchemename("dnet:review_levels");
+		return q;
+	}
+
 	private static List<ExternalReference> convertExternalRefs(OafProtos.Oaf oaf) {
 		ResultProtos.Result r = oaf.getEntity().getResult();
 		if (r.getExternalReferenceCount() > 0) {

dhp-workflows/dhp-aggregation/pom.xml

@@ -4,7 +4,7 @@
     <parent>
         <groupId>eu.dnetlib.dhp</groupId>
         <artifactId>dhp-workflows</artifactId>
-        <version>1.2.2-SNAPSHOT</version>
+        <version>1.2.3-SNAPSHOT</version>
     </parent>
     <artifactId>dhp-aggregation</artifactId>
     

dhp-workflows/dhp-aggregation/src/test/java/eu/dnetlib/dhp/collector/worker/DnetCollectorWorkerApplicationTests.java

@@ -8,6 +8,7 @@ import java.io.File;
 
 import org.junit.jupiter.api.AfterEach;
 import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Disabled;
 import org.junit.jupiter.api.Test;
 
 import com.fasterxml.jackson.databind.ObjectMapper;
@@ -19,6 +20,7 @@ import eu.dnetlib.dhp.collection.worker.utils.CollectorPluginFactory;
 import eu.dnetlib.message.Message;
 import eu.dnetlib.message.MessageManager;
 
+@Disabled
 public class DnetCollectorWorkerApplicationTests {
 
 	private final ArgumentApplicationParser argumentParser = mock(ArgumentApplicationParser.class);

dhp-workflows/dhp-blacklist/pom.xml

@@ -3,7 +3,7 @@
     <parent>
         <artifactId>dhp-workflows</artifactId>
         <groupId>eu.dnetlib.dhp</groupId>
-        <version>1.2.2-SNAPSHOT</version>
+        <version>1.2.3-SNAPSHOT</version>
     </parent>
     <modelVersion>4.0.0</modelVersion>
 

dhp-workflows/dhp-broker-events/pom.xml

@@ -1,11 +1,9 @@
 <?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/xsd/maven-4.0.0.xsd">
+<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/xsd/maven-4.0.0.xsd">
 	<parent>
 		<artifactId>dhp-workflows</artifactId>
 		<groupId>eu.dnetlib.dhp</groupId>
-		<version>1.2.2-SNAPSHOT</version>
+		<version>1.2.3-SNAPSHOT</version>
 	</parent>
 	<modelVersion>4.0.0</modelVersion>
 

dhp-workflows/dhp-broker-events/src/main/java/eu/dnetlib/dhp/broker/oa/GenerateEventsApplication.java

@@ -51,7 +51,7 @@ public class GenerateEventsApplication {
 			IOUtils
 				.toString(
 					GenerateEventsApplication.class
-						.getResourceAsStream("/eu/dnetlib/dhp/oa/graph/merge_claims_parameters.json")));
+						.getResourceAsStream("/eu/dnetlib/dhp/broker/oa/generate_broker_events.json")));
 		parser.parseArgument(args);
 
 		final Boolean isSparkSessionManaged = Optional
@@ -149,7 +149,8 @@ public class GenerateEventsApplication {
 		return r4;
 	}
 
-	private static <T, RT> Dataset<RT> relatedEntities(final Dataset<T> targets, final Dataset<Relation> rels,
+	private static <T, RT> Dataset<RT> relatedEntities(final Dataset<T> targets,
+		final Dataset<Relation> rels,
 		final Class<RT> clazz) {
 		return rels
 			.joinWith(targets, targets.col("id").equalTo(rels.col("target")), "inner")

dhp-workflows/dhp-broker-events/src/main/java/eu/dnetlib/dhp/broker/oa/matchers/UpdateMatcher.java

@@ -6,10 +6,14 @@ import java.util.Collection;
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
+import java.util.function.BiConsumer;
+import java.util.function.Function;
 
 import org.apache.commons.codec.digest.DigestUtils;
 import org.apache.commons.lang3.StringUtils;
 
+import eu.dnetlib.broker.objects.Publication;
+import eu.dnetlib.dhp.broker.model.Topic;
 import eu.dnetlib.dhp.broker.oa.util.UpdateInfo;
 import eu.dnetlib.dhp.broker.oa.util.aggregators.withRels.ResultWithRelations;
 import eu.dnetlib.dhp.schema.oaf.Field;
@@ -18,9 +22,17 @@ import eu.dnetlib.pace.config.DedupConfig;
 public abstract class UpdateMatcher<T> {
 
 	private final boolean multipleUpdate;
+	private final Function<T, Topic> topicFunction;
+	private final BiConsumer<Publication, T> compileHighlightFunction;
+	private final Function<T, String> highlightToStringFunction;
 
-	public UpdateMatcher(final boolean multipleUpdate) {
+	public UpdateMatcher(final boolean multipleUpdate, final Function<T, Topic> topicFunction,
+		final BiConsumer<Publication, T> compileHighlightFunction,
+		final Function<T, String> highlightToStringFunction) {
 		this.multipleUpdate = multipleUpdate;
+		this.topicFunction = topicFunction;
+		this.compileHighlightFunction = compileHighlightFunction;
+		this.highlightToStringFunction = highlightToStringFunction;
 	}
 
 	public Collection<UpdateInfo<T>> searchUpdatesForRecord(final ResultWithRelations res,
@@ -31,7 +43,11 @@ public abstract class UpdateMatcher<T> {
 
 		for (final ResultWithRelations source : others) {
 			if (source != res) {
-				for (final UpdateInfo<T> info : findUpdates(source, res, dedupConfig)) {
+				for (final T hl : findDifferences(source, res)) {
+					final Topic topic = getTopicFunction().apply(hl);
+					final UpdateInfo<T> info = new UpdateInfo<>(topic, hl, source, res, getCompileHighlightFunction(),
+						getHighlightToStringFunction(),
+						dedupConfig);
 					final String s = DigestUtils.md5Hex(info.getHighlightValueAsString());
 					if (!infoMap.containsKey(s) || infoMap.get(s).getTrust() < info.getTrust()) {
 					} else {
@@ -55,8 +71,7 @@ public abstract class UpdateMatcher<T> {
 		}
 	}
 
-	protected abstract List<UpdateInfo<T>> findUpdates(ResultWithRelations source, ResultWithRelations target,
-		DedupConfig dedupConfig);
+	protected abstract List<T> findDifferences(ResultWithRelations source, ResultWithRelations target);
 
 	protected static boolean isMissing(final List<Field<String>> list) {
 		return list == null || list.isEmpty() || StringUtils.isBlank(list.get(0).getValue());
@@ -66,4 +81,20 @@ public abstract class UpdateMatcher<T> {
 		return field == null || StringUtils.isBlank(field.getValue());
 	}
 
+	public boolean isMultipleUpdate() {
+		return multipleUpdate;
+	}
+
+	public Function<T, Topic> getTopicFunction() {
+		return topicFunction;
+	}
+
+	public BiConsumer<Publication, T> getCompileHighlightFunction() {
+		return compileHighlightFunction;
+	}
+
+	public Function<T, String> getHighlightToStringFunction() {
+		return highlightToStringFunction;
+	}
+
 }

dhp-workflows/dhp-broker-events/src/main/java/eu/dnetlib/dhp/broker/oa/matchers/relatedDatasets/AbstractEnrichMissingDataset.java

@@ -8,29 +8,26 @@ import java.util.stream.Collectors;
 import eu.dnetlib.dhp.broker.model.Topic;
 import eu.dnetlib.dhp.broker.oa.matchers.UpdateMatcher;
 import eu.dnetlib.dhp.broker.oa.util.ConversionUtils;
-import eu.dnetlib.dhp.broker.oa.util.UpdateInfo;
 import eu.dnetlib.dhp.broker.oa.util.aggregators.withRels.RelatedDataset;
 import eu.dnetlib.dhp.broker.oa.util.aggregators.withRels.ResultWithRelations;
 import eu.dnetlib.dhp.schema.oaf.Dataset;
-import eu.dnetlib.pace.config.DedupConfig;
 
 public abstract class AbstractEnrichMissingDataset
 	extends UpdateMatcher<eu.dnetlib.broker.objects.Dataset> {
 
-	private final Topic topic;
-
 	public AbstractEnrichMissingDataset(final Topic topic) {
-		super(true);
-		this.topic = topic;
+		super(true,
+			rel -> topic,
+			(p, rel) -> p.getDatasets().add(rel),
+			rel -> rel.getInstances().get(0).getUrl());
 	}
 
 	protected abstract boolean filterByType(String relType);
 
 	@Override
-	protected final List<UpdateInfo<eu.dnetlib.broker.objects.Dataset>> findUpdates(
+	protected final List<eu.dnetlib.broker.objects.Dataset> findDifferences(
 		final ResultWithRelations source,
-		final ResultWithRelations target,
-		final DedupConfig dedupConfig) {
+		final ResultWithRelations target) {
 
 		final Set<String> existingDatasets = target
 			.getDatasets()
@@ -47,26 +44,8 @@ public abstract class AbstractEnrichMissingDataset
 			.map(RelatedDataset::getRelDataset)
 			.filter(d -> !existingDatasets.contains(d.getId()))
 			.map(ConversionUtils::oafDatasetToBrokerDataset)
-			.map(i -> generateUpdateInfo(i, source, target, dedupConfig))
 			.collect(Collectors.toList());
 
 	}
 
-	protected final UpdateInfo<eu.dnetlib.broker.objects.Dataset> generateUpdateInfo(
-		final eu.dnetlib.broker.objects.Dataset highlightValue,
-		final ResultWithRelations source,
-		final ResultWithRelations target,
-		final DedupConfig dedupConfig) {
-		return new UpdateInfo<>(
-			getTopic(),
-			highlightValue, source, target,
-			(p, rel) -> p.getDatasets().add(rel),
-			rel -> rel.getInstances().get(0).getUrl(),
-			dedupConfig);
-	}
-
-	public Topic getTopic() {
-		return topic;
-	}
-
 }

dhp-workflows/dhp-broker-events/src/main/java/eu/dnetlib/dhp/broker/oa/matchers/relatedProjects/EnrichMissingProject.java

@@ -5,26 +5,25 @@ import java.util.Arrays;
 import java.util.List;
 import java.util.stream.Collectors;
 
+import eu.dnetlib.broker.objects.Project;
 import eu.dnetlib.dhp.broker.model.Topic;
 import eu.dnetlib.dhp.broker.oa.matchers.UpdateMatcher;
 import eu.dnetlib.dhp.broker.oa.util.ConversionUtils;
-import eu.dnetlib.dhp.broker.oa.util.UpdateInfo;
 import eu.dnetlib.dhp.broker.oa.util.aggregators.withRels.RelatedProject;
 import eu.dnetlib.dhp.broker.oa.util.aggregators.withRels.ResultWithRelations;
-import eu.dnetlib.pace.config.DedupConfig;
 
 public class EnrichMissingProject
 	extends UpdateMatcher<eu.dnetlib.broker.objects.Project> {
 
 	public EnrichMissingProject() {
-		super(true);
+		super(true,
+			prj -> Topic.ENRICH_MISSING_PROJECT,
+			(p, prj) -> p.getProjects().add(prj),
+			prj -> prj.getFunder() + "::" + prj.getFundingProgram() + prj.getCode());
 	}
 
 	@Override
-	protected List<UpdateInfo<eu.dnetlib.broker.objects.Project>> findUpdates(final ResultWithRelations source,
-		final ResultWithRelations target,
-		final DedupConfig dedupConfig) {
-
+	protected List<Project> findDifferences(final ResultWithRelations source, final ResultWithRelations target) {
 		if (source.getProjects().isEmpty()) {
 			return Arrays.asList();
 		} else {
@@ -33,21 +32,7 @@ public class EnrichMissingProject
 				.stream()
 				.map(RelatedProject::getRelProject)
 				.map(ConversionUtils::oafProjectToBrokerProject)
-				.map(p -> generateUpdateInfo(p, source, target, dedupConfig))
 				.collect(Collectors.toList());
 		}
 	}
-
-	public UpdateInfo<eu.dnetlib.broker.objects.Project> generateUpdateInfo(
-		final eu.dnetlib.broker.objects.Project highlightValue,
-		final ResultWithRelations source,
-		final ResultWithRelations target,
-		final DedupConfig dedupConfig) {
-		return new UpdateInfo<>(
-			Topic.ENRICH_MISSING_PROJECT,
-			highlightValue, source, target,
-			(p, prj) -> p.getProjects().add(prj),
-			prj -> prj.getFunder() + "::" + prj.getFundingProgram() + prj.getCode(), dedupConfig);
-	}
-
 }

dhp-workflows/dhp-broker-events/src/main/java/eu/dnetlib/dhp/broker/oa/matchers/relatedProjects/EnrichMoreProject.java

@@ -8,22 +8,22 @@ import java.util.stream.Collectors;
 import eu.dnetlib.dhp.broker.model.Topic;
 import eu.dnetlib.dhp.broker.oa.matchers.UpdateMatcher;
 import eu.dnetlib.dhp.broker.oa.util.ConversionUtils;
-import eu.dnetlib.dhp.broker.oa.util.UpdateInfo;
 import eu.dnetlib.dhp.broker.oa.util.aggregators.withRels.RelatedProject;
 import eu.dnetlib.dhp.broker.oa.util.aggregators.withRels.ResultWithRelations;
 import eu.dnetlib.dhp.schema.oaf.Project;
-import eu.dnetlib.pace.config.DedupConfig;
 
 public class EnrichMoreProject extends UpdateMatcher<eu.dnetlib.broker.objects.Project> {
 
 	public EnrichMoreProject() {
-		super(true);
+		super(true,
+			prj -> Topic.ENRICH_MORE_PROJECT,
+			(p, prj) -> p.getProjects().add(prj),
+			prj -> prj.getFunder() + "::" + prj.getFundingProgram() + prj.getCode());
 	}
 
 	@Override
-	protected List<UpdateInfo<eu.dnetlib.broker.objects.Project>> findUpdates(final ResultWithRelations source,
-		final ResultWithRelations target,
-		final DedupConfig dedupConfig) {
+	protected List<eu.dnetlib.broker.objects.Project> findDifferences(final ResultWithRelations source,
+		final ResultWithRelations target) {
 
 		final Set<String> existingProjects = source
 			.getProjects()
@@ -38,20 +38,7 @@ public class EnrichMoreProject extends UpdateMatcher<eu.dnetlib.broker.objects.P
 			.map(RelatedProject::getRelProject)
 			.filter(p -> !existingProjects.contains(p.getId()))
 			.map(ConversionUtils::oafProjectToBrokerProject)
-			.map(p -> generateUpdateInfo(p, source, target, dedupConfig))
 			.collect(Collectors.toList());
 	}
 
-	public UpdateInfo<eu.dnetlib.broker.objects.Project> generateUpdateInfo(
-		final eu.dnetlib.broker.objects.Project highlightValue,
-		final ResultWithRelations source,
-		final ResultWithRelations target,
-		final DedupConfig dedupConfig) {
-		return new UpdateInfo<>(
-			Topic.ENRICH_MORE_PROJECT,
-			highlightValue, source, target,
-			(p, prj) -> p.getProjects().add(prj),
-			prj -> prj.getFunder() + "::" + prj.getFundingProgram() + prj.getCode(), dedupConfig);
-	}
-
 }

dhp-workflows/dhp-broker-events/src/main/java/eu/dnetlib/dhp/broker/oa/matchers/relatedPublications/AbstractEnrichMissingPublication.java

@@ -8,29 +8,27 @@ import java.util.stream.Collectors;
 import eu.dnetlib.dhp.broker.model.Topic;
 import eu.dnetlib.dhp.broker.oa.matchers.UpdateMatcher;
 import eu.dnetlib.dhp.broker.oa.util.ConversionUtils;
-import eu.dnetlib.dhp.broker.oa.util.UpdateInfo;
 import eu.dnetlib.dhp.broker.oa.util.aggregators.withRels.RelatedPublication;
 import eu.dnetlib.dhp.broker.oa.util.aggregators.withRels.ResultWithRelations;
 import eu.dnetlib.dhp.schema.oaf.Publication;
-import eu.dnetlib.pace.config.DedupConfig;
 
 public abstract class AbstractEnrichMissingPublication
 	extends UpdateMatcher<eu.dnetlib.broker.objects.Publication> {
 
-	private final Topic topic;
-
 	public AbstractEnrichMissingPublication(final Topic topic) {
-		super(true);
-		this.topic = topic;
+		super(true,
+			rel -> topic,
+			(p, rel) -> p.getPublications().add(rel),
+			rel -> rel.getInstances().get(0).getUrl());
+
 	}
 
 	protected abstract boolean filterByType(String relType);
 
 	@Override
-	protected final List<UpdateInfo<eu.dnetlib.broker.objects.Publication>> findUpdates(
+	protected final List<eu.dnetlib.broker.objects.Publication> findDifferences(
 		final ResultWithRelations source,
-		final ResultWithRelations target,
-		final DedupConfig dedupConfig) {
+		final ResultWithRelations target) {
 
 		final Set<String> existingPublications = target
 			.getPublications()
@@ -47,24 +45,7 @@ public abstract class AbstractEnrichMissingPublication
 			.map(RelatedPublication::getRelPublication)
 			.filter(d -> !existingPublications.contains(d.getId()))
 			.map(ConversionUtils::oafResultToBrokerPublication)
-			.map(i -> generateUpdateInfo(i, source, target, dedupConfig))
 			.collect(Collectors.toList());
-
 	}
 
-	protected final UpdateInfo<eu.dnetlib.broker.objects.Publication> generateUpdateInfo(
-		final eu.dnetlib.broker.objects.Publication highlightValue,
-		final ResultWithRelations source,
-		final ResultWithRelations target,
-		final DedupConfig dedupConfig) {
-		return new UpdateInfo<>(
-			getTopic(),
-			highlightValue, source, target,
-			(p, rel) -> p.getPublications().add(rel),
-			rel -> rel.getInstances().get(0).getUrl(), dedupConfig);
-	}
-
-	public Topic getTopic() {
-		return topic;
-	}
 }

dhp-workflows/dhp-broker-events/src/main/java/eu/dnetlib/dhp/broker/oa/matchers/relatedSoftware/EnrichMissingSoftware.java

@@ -8,23 +8,23 @@ import java.util.stream.Collectors;
 import eu.dnetlib.dhp.broker.model.Topic;
 import eu.dnetlib.dhp.broker.oa.matchers.UpdateMatcher;
 import eu.dnetlib.dhp.broker.oa.util.ConversionUtils;
-import eu.dnetlib.dhp.broker.oa.util.UpdateInfo;
 import eu.dnetlib.dhp.broker.oa.util.aggregators.withRels.RelatedSoftware;
 import eu.dnetlib.dhp.broker.oa.util.aggregators.withRels.ResultWithRelations;
-import eu.dnetlib.pace.config.DedupConfig;
 
 public class EnrichMissingSoftware
 	extends UpdateMatcher<eu.dnetlib.broker.objects.Software> {
 
 	public EnrichMissingSoftware() {
-		super(true);
+		super(true,
+			s -> Topic.ENRICH_MISSING_SOFTWARE,
+			(p, s) -> p.getSoftwares().add(s),
+			s -> s.getName());
 	}
 
 	@Override
-	protected List<UpdateInfo<eu.dnetlib.broker.objects.Software>> findUpdates(
+	protected List<eu.dnetlib.broker.objects.Software> findDifferences(
 		final ResultWithRelations source,
-		final ResultWithRelations target,
-		final DedupConfig dedupConfig) {
+		final ResultWithRelations target) {
 
 		if (source.getSoftwares().isEmpty()) {
 			return Arrays.asList();
@@ -34,21 +34,8 @@ public class EnrichMissingSoftware
 				.stream()
 				.map(RelatedSoftware::getRelSoftware)
 				.map(ConversionUtils::oafSoftwareToBrokerSoftware)
-				.map(p -> generateUpdateInfo(p, source, target, dedupConfig))
 				.collect(Collectors.toList());
 		}
 	}
 
-	public UpdateInfo<eu.dnetlib.broker.objects.Software> generateUpdateInfo(
-		final eu.dnetlib.broker.objects.Software highlightValue,
-		final ResultWithRelations source,
-		final ResultWithRelations target,
-		final DedupConfig dedupConfig) {
-		return new UpdateInfo<>(
-			Topic.ENRICH_MISSING_SOFTWARE,
-			highlightValue, source, target,
-			(p, s) -> p.getSoftwares().add(s),
-			s -> s.getName(), dedupConfig);
-	}
-
 }

dhp-workflows/dhp-broker-events/src/main/java/eu/dnetlib/dhp/broker/oa/matchers/relatedSoftware/EnrichMoreSoftware.java

@@ -8,24 +8,24 @@ import java.util.stream.Collectors;
 import eu.dnetlib.dhp.broker.model.Topic;
 import eu.dnetlib.dhp.broker.oa.matchers.UpdateMatcher;
 import eu.dnetlib.dhp.broker.oa.util.ConversionUtils;
-import eu.dnetlib.dhp.broker.oa.util.UpdateInfo;
 import eu.dnetlib.dhp.broker.oa.util.aggregators.withRels.RelatedSoftware;
 import eu.dnetlib.dhp.broker.oa.util.aggregators.withRels.ResultWithRelations;
 import eu.dnetlib.dhp.schema.oaf.Software;
-import eu.dnetlib.pace.config.DedupConfig;
 
 public class EnrichMoreSoftware
 	extends UpdateMatcher<eu.dnetlib.broker.objects.Software> {
 
 	public EnrichMoreSoftware() {
-		super(true);
+		super(true,
+			s -> Topic.ENRICH_MORE_SOFTWARE,
+			(p, s) -> p.getSoftwares().add(s),
+			s -> s.getName());
 	}
 
 	@Override
-	protected List<UpdateInfo<eu.dnetlib.broker.objects.Software>> findUpdates(
+	protected List<eu.dnetlib.broker.objects.Software> findDifferences(
 		final ResultWithRelations source,
-		final ResultWithRelations target,
-		final DedupConfig dedupConfig) {
+		final ResultWithRelations target) {
 
 		final Set<String> existingSoftwares = source
 			.getSoftwares()
@@ -40,20 +40,7 @@ public class EnrichMoreSoftware
 			.map(RelatedSoftware::getRelSoftware)
 			.filter(p -> !existingSoftwares.contains(p.getId()))
 			.map(ConversionUtils::oafSoftwareToBrokerSoftware)
-			.map(p -> generateUpdateInfo(p, source, target, dedupConfig))
 			.collect(Collectors.toList());
 	}
 
-	public UpdateInfo<eu.dnetlib.broker.objects.Software> generateUpdateInfo(
-		final eu.dnetlib.broker.objects.Software highlightValue,
-		final ResultWithRelations source,
-		final ResultWithRelations target,
-		final DedupConfig dedupConfig) {
-		return new UpdateInfo<>(
-			Topic.ENRICH_MORE_SOFTWARE,
-			highlightValue, source, target,
-			(p, s) -> p.getSoftwares().add(s),
-			s -> s.getName(), dedupConfig);
-	}
-
 }

dhp-workflows/dhp-broker-events/src/main/java/eu/dnetlib/dhp/broker/oa/matchers/simple/EnrichMissingAbstract.java

@@ -7,38 +7,25 @@ import java.util.List;
 
 import eu.dnetlib.dhp.broker.model.Topic;
 import eu.dnetlib.dhp.broker.oa.matchers.UpdateMatcher;
-import eu.dnetlib.dhp.broker.oa.util.UpdateInfo;
 import eu.dnetlib.dhp.broker.oa.util.aggregators.withRels.ResultWithRelations;
-import eu.dnetlib.pace.config.DedupConfig;
 
 public class EnrichMissingAbstract extends UpdateMatcher<String> {
 
 	public EnrichMissingAbstract() {
-		super(false);
+		super(false,
+			s -> Topic.ENRICH_MISSING_ABSTRACT,
+			(p, s) -> p.getAbstracts().add(s),
+			s -> s);
 	}
 
 	@Override
-	protected List<UpdateInfo<String>> findUpdates(final ResultWithRelations source,
-		final ResultWithRelations target,
-		final DedupConfig dedupConfig) {
+	protected List<String> findDifferences(final ResultWithRelations source,
+		final ResultWithRelations target) {
 		if (isMissing(target.getResult().getDescription()) && !isMissing(source.getResult().getDescription())) {
 			return Arrays
-				.asList(
-					generateUpdateInfo(
-						source.getResult().getDescription().get(0).getValue(), source, target, dedupConfig));
+				.asList(source.getResult().getDescription().get(0).getValue());
 		}
 		return new ArrayList<>();
 	}
 
-	public UpdateInfo<String> generateUpdateInfo(final String highlightValue,
-		final ResultWithRelations source,
-		final ResultWithRelations target,
-		final DedupConfig dedupConfig) {
-		return new UpdateInfo<>(
-			Topic.ENRICH_MISSING_ABSTRACT,
-			highlightValue, source, target,
-			(p, s) -> p.getAbstracts().add(s),
-			s -> s, dedupConfig);
-	}
-
 }

dhp-workflows/dhp-broker-events/src/main/java/eu/dnetlib/dhp/broker/oa/matchers/simple/EnrichMissingAuthorOrcid.java

@@ -8,22 +8,22 @@ import java.util.stream.Collectors;
 
 import eu.dnetlib.dhp.broker.model.Topic;
 import eu.dnetlib.dhp.broker.oa.matchers.UpdateMatcher;
-import eu.dnetlib.dhp.broker.oa.util.UpdateInfo;
 import eu.dnetlib.dhp.broker.oa.util.aggregators.withRels.ResultWithRelations;
 import eu.dnetlib.dhp.schema.oaf.Author;
 import eu.dnetlib.dhp.schema.oaf.StructuredProperty;
-import eu.dnetlib.pace.config.DedupConfig;
 
 public class EnrichMissingAuthorOrcid extends UpdateMatcher<String> {
 
 	public EnrichMissingAuthorOrcid() {
-		super(true);
+		super(true,
+			aut -> Topic.ENRICH_MISSING_AUTHOR_ORCID,
+			(p, aut) -> p.getCreators().add(aut),
+			aut -> aut);
 	}
 
 	@Override
-	protected List<UpdateInfo<String>> findUpdates(final ResultWithRelations source,
-		final ResultWithRelations target,
-		final DedupConfig dedupConfig) {
+	protected List<String> findDifferences(final ResultWithRelations source,
+		final ResultWithRelations target) {
 
 		final Set<String> existingOrcids = target
 			.getResult()
@@ -35,7 +35,7 @@ public class EnrichMissingAuthorOrcid extends UpdateMatcher<String> {
 			.map(pid -> pid.getValue())
 			.collect(Collectors.toSet());
 
-		final List<UpdateInfo<String>> list = new ArrayList<>();
+		final List<String> list = new ArrayList<>();
 
 		for (final Author author : source.getResult().getAuthor()) {
 			final String name = author.getFullname();
@@ -43,26 +43,11 @@ public class EnrichMissingAuthorOrcid extends UpdateMatcher<String> {
 			for (final StructuredProperty pid : author.getPid()) {
 				if (pid.getQualifier().getClassid().equalsIgnoreCase("orcid")
 					&& !existingOrcids.contains(pid.getValue())) {
-					list
-						.add(
-							generateUpdateInfo(name + " [ORCID: " + pid.getValue() + "]", source, target, dedupConfig));
-					;
+					list.add(name + " [ORCID: " + pid.getValue() + "]");
 				}
 			}
 		}
 
 		return list;
 	}
-
-	public UpdateInfo<String> generateUpdateInfo(final String highlightValue,
-		final ResultWithRelations source,
-		final ResultWithRelations target,
-		final DedupConfig dedupConfig) {
-		return new UpdateInfo<>(
-			Topic.ENRICH_MISSING_AUTHOR_ORCID,
-			highlightValue, source, target,
-			(p, aut) -> p.getCreators().add(aut),
-			aut -> aut,
-			dedupConfig);
-	}
 }

dhp-workflows/dhp-broker-events/src/main/java/eu/dnetlib/dhp/broker/oa/matchers/simple/EnrichMissingOpenAccess.java

@@ -10,20 +10,20 @@ import eu.dnetlib.dhp.broker.model.Topic;
 import eu.dnetlib.dhp.broker.oa.matchers.UpdateMatcher;
 import eu.dnetlib.dhp.broker.oa.util.BrokerConstants;
 import eu.dnetlib.dhp.broker.oa.util.ConversionUtils;
-import eu.dnetlib.dhp.broker.oa.util.UpdateInfo;
 import eu.dnetlib.dhp.broker.oa.util.aggregators.withRels.ResultWithRelations;
-import eu.dnetlib.pace.config.DedupConfig;
 
 public class EnrichMissingOpenAccess extends UpdateMatcher<Instance> {
 
 	public EnrichMissingOpenAccess() {
-		super(true);
+		super(true,
+			i -> Topic.ENRICH_MISSING_OA_VERSION,
+			(p, i) -> p.getInstances().add(i),
+			Instance::getUrl);
 	}
 
 	@Override
-	protected List<UpdateInfo<Instance>> findUpdates(final ResultWithRelations source,
-		final ResultWithRelations target,
-		final DedupConfig dedupConfig) {
+	protected List<Instance> findDifferences(final ResultWithRelations source,
+		final ResultWithRelations target) {
 		final long count = target
 			.getResult()
 			.getInstance()
@@ -43,19 +43,7 @@ public class EnrichMissingOpenAccess extends UpdateMatcher<Instance> {
 			.filter(i -> i.getAccessright().getClassid().equals(BrokerConstants.OPEN_ACCESS))
 			.map(ConversionUtils::oafInstanceToBrokerInstances)
 			.flatMap(List::stream)
-			.map(i -> generateUpdateInfo(i, source, target, dedupConfig))
 			.collect(Collectors.toList());
 	}
 
-	public UpdateInfo<Instance> generateUpdateInfo(final Instance highlightValue,
-		final ResultWithRelations source,
-		final ResultWithRelations target,
-		final DedupConfig dedupConfig) {
-		return new UpdateInfo<>(
-			Topic.ENRICH_MISSING_OA_VERSION,
-			highlightValue, source, target,
-			(p, i) -> p.getInstances().add(i),
-			Instance::getUrl, dedupConfig);
-	}
-
 }

dhp-workflows/dhp-broker-events/src/main/java/eu/dnetlib/dhp/broker/oa/matchers/simple/EnrichMissingPid.java

@@ -9,20 +9,20 @@ import eu.dnetlib.broker.objects.Pid;
 import eu.dnetlib.dhp.broker.model.Topic;
 import eu.dnetlib.dhp.broker.oa.matchers.UpdateMatcher;
 import eu.dnetlib.dhp.broker.oa.util.ConversionUtils;
-import eu.dnetlib.dhp.broker.oa.util.UpdateInfo;
 import eu.dnetlib.dhp.broker.oa.util.aggregators.withRels.ResultWithRelations;
-import eu.dnetlib.pace.config.DedupConfig;
 
 public class EnrichMissingPid extends UpdateMatcher<Pid> {
 
 	public EnrichMissingPid() {
-		super(true);
+		super(true,
+			pid -> Topic.ENRICH_MISSING_PID,
+			(p, pid) -> p.getPids().add(pid),
+			pid -> pid.getType() + "::" + pid.getValue());
 	}
 
 	@Override
-	protected List<UpdateInfo<Pid>> findUpdates(final ResultWithRelations source,
-		final ResultWithRelations target,
-		final DedupConfig dedupConfig) {
+	protected List<Pid> findDifferences(final ResultWithRelations source,
+		final ResultWithRelations target) {
 		final long count = target.getResult().getPid().size();
 
 		if (count > 0) {
@@ -34,19 +34,7 @@ public class EnrichMissingPid extends UpdateMatcher<Pid> {
 			.getPid()
 			.stream()
 			.map(ConversionUtils::oafPidToBrokerPid)
-			.map(i -> generateUpdateInfo(i, source, target, dedupConfig))
 			.collect(Collectors.toList());
 	}
 
-	public UpdateInfo<Pid> generateUpdateInfo(final Pid highlightValue,
-		final ResultWithRelations source,
-		final ResultWithRelations target,
-		final DedupConfig dedupConfig) {
-		return new UpdateInfo<>(
-			Topic.ENRICH_MISSING_PID,
-			highlightValue, source, target,
-			(p, pid) -> p.getPids().add(pid),
-			pid -> pid.getType() + "::" + pid.getValue(), dedupConfig);
-	}
-
 }

dhp-workflows/dhp-broker-events/src/main/java/eu/dnetlib/dhp/broker/oa/matchers/simple/EnrichMissingPublicationDate.java

@@ -7,39 +7,25 @@ import java.util.List;
 
 import eu.dnetlib.dhp.broker.model.Topic;
 import eu.dnetlib.dhp.broker.oa.matchers.UpdateMatcher;
-import eu.dnetlib.dhp.broker.oa.util.UpdateInfo;
 import eu.dnetlib.dhp.broker.oa.util.aggregators.withRels.ResultWithRelations;
-import eu.dnetlib.pace.config.DedupConfig;
 
 public class EnrichMissingPublicationDate extends UpdateMatcher<String> {
 
 	public EnrichMissingPublicationDate() {
-		super(false);
+		super(false,
+			date -> Topic.ENRICH_MISSING_PUBLICATION_DATE,
+			(p, date) -> p.setPublicationdate(date),
+			s -> s);
 	}
 
 	@Override
-	protected List<UpdateInfo<String>> findUpdates(final ResultWithRelations source,
-		final ResultWithRelations target,
-		final DedupConfig dedupConfig) {
+	protected List<String> findDifferences(final ResultWithRelations source,
+		final ResultWithRelations target) {
 		if (isMissing(target.getResult().getDateofacceptance())
 			&& !isMissing(source.getResult().getDateofacceptance())) {
-			return Arrays
-				.asList(
-					generateUpdateInfo(
-						source.getResult().getDateofacceptance().getValue(), source, target, dedupConfig));
+			return Arrays.asList(source.getResult().getDateofacceptance().getValue());
 		}
 		return new ArrayList<>();
 	}
 
-	public UpdateInfo<String> generateUpdateInfo(final String highlightValue,
-		final ResultWithRelations source,
-		final ResultWithRelations target,
-		final DedupConfig dedupConfig) {
-		return new UpdateInfo<>(
-			Topic.ENRICH_MISSING_PUBLICATION_DATE,
-			highlightValue, source, target,
-			(p, date) -> p.setPublicationdate(date),
-			s -> s, dedupConfig);
-	}
-
 }

dhp-workflows/dhp-broker-events/src/main/java/eu/dnetlib/dhp/broker/oa/matchers/simple/EnrichMissingSubject.java

@@ -10,22 +10,22 @@ import org.apache.commons.lang3.tuple.Pair;
 import eu.dnetlib.dhp.broker.model.Topic;
 import eu.dnetlib.dhp.broker.oa.matchers.UpdateMatcher;
 import eu.dnetlib.dhp.broker.oa.util.ConversionUtils;
-import eu.dnetlib.dhp.broker.oa.util.UpdateInfo;
 import eu.dnetlib.dhp.broker.oa.util.aggregators.withRels.ResultWithRelations;
 import eu.dnetlib.dhp.schema.oaf.Qualifier;
 import eu.dnetlib.dhp.schema.oaf.StructuredProperty;
-import eu.dnetlib.pace.config.DedupConfig;
 
 public class EnrichMissingSubject extends UpdateMatcher<Pair<String, String>> {
 
 	public EnrichMissingSubject() {
-		super(true);
+		super(true,
+			pair -> Topic.fromPath("ENRICH/MISSING/SUBJECT/" + pair.getLeft()),
+			(p, pair) -> p.getSubjects().add(pair.getRight()),
+			pair -> pair.getLeft() + "::" + pair.getRight());
 	}
 
 	@Override
-	protected List<UpdateInfo<Pair<String, String>>> findUpdates(final ResultWithRelations source,
-		final ResultWithRelations target,
-		final DedupConfig dedupConfig) {
+	protected List<Pair<String, String>> findDifferences(final ResultWithRelations source,
+		final ResultWithRelations target) {
 		final Set<String> existingTypes = target
 			.getResult()
 			.getSubject()
@@ -40,20 +40,7 @@ public class EnrichMissingSubject extends UpdateMatcher<Pair<String, String>> {
 			.stream()
 			.filter(pid -> !existingTypes.contains(pid.getQualifier().getClassid()))
 			.map(ConversionUtils::oafSubjectToPair)
-			.map(i -> generateUpdateInfo(i, source, target, dedupConfig))
 			.collect(Collectors.toList());
 	}
 
-	public UpdateInfo<Pair<String, String>> generateUpdateInfo(final Pair<String, String> highlightValue,
-		final ResultWithRelations source,
-		final ResultWithRelations target,
-		final DedupConfig dedupConfig) {
-
-		return new UpdateInfo<>(
-			Topic.fromPath("ENRICH/MISSING/SUBJECT/" + highlightValue.getLeft()),
-			highlightValue, source, target,
-			(p, pair) -> p.getSubjects().add(pair.getRight()),
-			pair -> pair.getLeft() + "::" + pair.getRight(), dedupConfig);
-	}
-
 }

dhp-workflows/dhp-broker-events/src/main/java/eu/dnetlib/dhp/broker/oa/matchers/simple/EnrichMoreOpenAccess.java

@@ -10,20 +10,20 @@ import eu.dnetlib.dhp.broker.model.Topic;
 import eu.dnetlib.dhp.broker.oa.matchers.UpdateMatcher;
 import eu.dnetlib.dhp.broker.oa.util.BrokerConstants;
 import eu.dnetlib.dhp.broker.oa.util.ConversionUtils;
-import eu.dnetlib.dhp.broker.oa.util.UpdateInfo;
 import eu.dnetlib.dhp.broker.oa.util.aggregators.withRels.ResultWithRelations;
-import eu.dnetlib.pace.config.DedupConfig;
 
 public class EnrichMoreOpenAccess extends UpdateMatcher<Instance> {
 
 	public EnrichMoreOpenAccess() {
-		super(true);
+		super(true,
+			i -> Topic.ENRICH_MORE_OA_VERSION,
+			(p, i) -> p.getInstances().add(i),
+			Instance::getUrl);
 	}
 
 	@Override
-	protected List<UpdateInfo<Instance>> findUpdates(final ResultWithRelations source,
-		final ResultWithRelations target,
-		final DedupConfig dedupConfig) {
+	protected List<Instance> findDifferences(final ResultWithRelations source,
+		final ResultWithRelations target) {
 		final Set<String> urls = target
 			.getResult()
 			.getInstance()
@@ -41,19 +41,7 @@ public class EnrichMoreOpenAccess extends UpdateMatcher<Instance> {
 			.map(ConversionUtils::oafInstanceToBrokerInstances)
 			.flatMap(List::stream)
 			.filter(i -> !urls.contains(i.getUrl()))
-			.map(i -> generateUpdateInfo(i, source, target, dedupConfig))
 			.collect(Collectors.toList());
 	}
 
-	public UpdateInfo<Instance> generateUpdateInfo(final Instance highlightValue,
-		final ResultWithRelations source,
-		final ResultWithRelations target,
-		final DedupConfig dedupConfig) {
-		return new UpdateInfo<>(
-			Topic.ENRICH_MORE_OA_VERSION,
-			highlightValue, source, target,
-			(p, i) -> p.getInstances().add(i),
-			Instance::getUrl, dedupConfig);
-	}
-
 }

dhp-workflows/dhp-broker-events/src/main/java/eu/dnetlib/dhp/broker/oa/matchers/simple/EnrichMorePid.java

@@ -9,20 +9,20 @@ import eu.dnetlib.broker.objects.Pid;
 import eu.dnetlib.dhp.broker.model.Topic;
 import eu.dnetlib.dhp.broker.oa.matchers.UpdateMatcher;
 import eu.dnetlib.dhp.broker.oa.util.ConversionUtils;
-import eu.dnetlib.dhp.broker.oa.util.UpdateInfo;
 import eu.dnetlib.dhp.broker.oa.util.aggregators.withRels.ResultWithRelations;
-import eu.dnetlib.pace.config.DedupConfig;
 
 public class EnrichMorePid extends UpdateMatcher<Pid> {
 
 	public EnrichMorePid() {
-		super(true);
+		super(true,
+			pid -> Topic.ENRICH_MORE_PID,
+			(p, pid) -> p.getPids().add(pid),
+			pid -> pid.getType() + "::" + pid.getValue());
 	}
 
 	@Override
-	protected List<UpdateInfo<Pid>> findUpdates(final ResultWithRelations source,
-		final ResultWithRelations target,
-		final DedupConfig dedupConfig) {
+	protected List<Pid> findDifferences(final ResultWithRelations source,
+		final ResultWithRelations target) {
 		final Set<String> existingPids = target
 			.getResult()
 			.getPid()
@@ -36,19 +36,7 @@ public class EnrichMorePid extends UpdateMatcher<Pid> {
 			.stream()
 			.filter(pid -> !existingPids.contains(pid.getQualifier().getClassid() + "::" + pid.getValue()))
 			.map(ConversionUtils::oafPidToBrokerPid)
-			.map(i -> generateUpdateInfo(i, source, target, dedupConfig))
 			.collect(Collectors.toList());
 	}
 
-	public UpdateInfo<Pid> generateUpdateInfo(final Pid highlightValue,
-		final ResultWithRelations source,
-		final ResultWithRelations target,
-		final DedupConfig dedupConfig) {
-		return new UpdateInfo<>(
-			Topic.ENRICH_MORE_PID,
-			highlightValue, source, target,
-			(p, pid) -> p.getPids().add(pid),
-			pid -> pid.getType() + "::" + pid.getValue(), dedupConfig);
-	}
-
 }

dhp-workflows/dhp-broker-events/src/main/java/eu/dnetlib/dhp/broker/oa/matchers/simple/EnrichMoreSubject.java

@@ -10,20 +10,20 @@ import org.apache.commons.lang3.tuple.Pair;
 import eu.dnetlib.dhp.broker.model.Topic;
 import eu.dnetlib.dhp.broker.oa.matchers.UpdateMatcher;
 import eu.dnetlib.dhp.broker.oa.util.ConversionUtils;
-import eu.dnetlib.dhp.broker.oa.util.UpdateInfo;
 import eu.dnetlib.dhp.broker.oa.util.aggregators.withRels.ResultWithRelations;
-import eu.dnetlib.pace.config.DedupConfig;
 
 public class EnrichMoreSubject extends UpdateMatcher<Pair<String, String>> {
 
 	public EnrichMoreSubject() {
-		super(true);
+		super(true,
+			pair -> Topic.fromPath("ENRICH/MORE/SUBJECT/" + pair.getLeft()),
+			(p, pair) -> p.getSubjects().add(pair.getRight()),
+			pair -> pair.getLeft() + "::" + pair.getRight());
 	}
 
 	@Override
-	protected List<UpdateInfo<Pair<String, String>>> findUpdates(final ResultWithRelations source,
-		final ResultWithRelations target,
-		final DedupConfig dedupConfig) {
+	protected List<Pair<String, String>> findDifferences(final ResultWithRelations source,
+		final ResultWithRelations target) {
 		final Set<String> existingSubjects = target
 			.getResult()
 			.getSubject()
@@ -37,20 +37,7 @@ public class EnrichMoreSubject extends UpdateMatcher<Pair<String, String>> {
 			.stream()
 			.filter(pid -> !existingSubjects.contains(pid.getQualifier().getClassid() + "::" + pid.getValue()))
 			.map(ConversionUtils::oafSubjectToPair)
-			.map(i -> generateUpdateInfo(i, source, target, dedupConfig))
 			.collect(Collectors.toList());
 	}
 
-	public UpdateInfo<Pair<String, String>> generateUpdateInfo(final Pair<String, String> highlightValue,
-		final ResultWithRelations source,
-		final ResultWithRelations target,
-		final DedupConfig dedupConfig) {
-
-		return new UpdateInfo<>(
-			Topic.fromPath("ENRICH/MORE/SUBJECT/" + highlightValue.getLeft()),
-			highlightValue, source, target,
-			(p, pair) -> p.getSubjects().add(pair.getRight()),
-			pair -> pair.getLeft() + "::" + pair.getRight(), dedupConfig);
-	}
-
 }

dhp-workflows/dhp-broker-events/src/main/resources/eu/dnetlib/dhp/broker/oa/generate_all/oozie_app/config-default.xml

@@ -0,0 +1,18 @@
+<configuration>
+    <property>
+        <name>jobTracker</name>
+        <value>yarnRM</value>
+    </property>
+    <property>
+        <name>nameNode</name>
+        <value>hdfs://nameservice1</value>
+    </property>
+    <property>
+        <name>oozie.use.system.libpath</name>
+        <value>true</value>
+    </property>
+    <property>
+        <name>oozie.action.sharelib.for.spark</name>
+        <value>spark2</value>
+    </property>
+</configuration>

dhp-workflows/dhp-broker-events/src/main/resources/eu/dnetlib/dhp/broker/oa/generate_all/oozie_app/workflow.xml

@@ -0,0 +1,99 @@
+<workflow-app name="create broker events" xmlns="uri:oozie:workflow:0.5">
+
+    <parameters>
+        <property>
+            <name>graphInputPath</name>
+            <description>the path where the graph is stored</description>
+        </property>
+        <property>
+            <name>eventsOutputPath</name>
+            <description>the path where the the events will be stored</description>
+        </property>
+        <property>
+            <name>isLookupUrl</name>
+            <description>the address of the lookUp service</description>
+        </property>
+        <property>
+            <name>dedupConfProfId</name>
+            <description>the id of a valid Dedup Configuration Profile</description>
+        </property>
+
+        <property>
+            <name>sparkDriverMemory</name>
+            <description>memory for driver process</description>
+        </property>
+        <property>
+            <name>sparkExecutorMemory</name>
+            <description>memory for individual executor</description>
+        </property>
+        <property>
+            <name>sparkExecutorCores</name>
+            <description>number of cores used by single executor</description>
+        </property>
+        <property>
+            <name>oozieActionShareLibForSpark2</name>
+            <description>oozie action sharelib for spark 2.*</description>
+        </property>
+        <property>
+            <name>spark2ExtraListeners</name>
+            <value>com.cloudera.spark.lineage.NavigatorAppListener</value>
+            <description>spark 2.* extra listeners classname</description>
+        </property>
+        <property>
+            <name>spark2SqlQueryExecutionListeners</name>
+            <value>com.cloudera.spark.lineage.NavigatorQueryListener</value>
+            <description>spark 2.* sql query execution listeners classname</description>
+        </property>
+        <property>
+            <name>spark2YarnHistoryServerAddress</name>
+            <description>spark 2.* yarn history server address</description>
+        </property>
+        <property>
+            <name>spark2EventLogDir</name>
+            <description>spark 2.* event log dir location</description>
+        </property>
+    </parameters>
+
+    <global>
+        <job-tracker>${jobTracker}</job-tracker>
+        <name-node>${nameNode}</name-node>
+        <configuration>
+            <property>
+                <name>mapreduce.job.queuename</name>
+                <value>${queueName}</value>
+            </property>
+            <property>
+                <name>oozie.launcher.mapred.job.queue.name</name>
+                <value>${oozieLauncherQueueName}</value>
+            </property>
+            <property>
+                <name>oozie.action.sharelib.for.spark</name>
+                <value>${oozieActionShareLibForSpark2}</value>
+            </property>
+        </configuration>
+    </global>
+
+    <start to="generate_events"/>
+
+    <kill name="Kill">
+        <message>Action failed, error message[${wf:errorMessage(wf:lastErrorNode())}]</message>
+    </kill>
+
+    <action name="generate_events">
+        <java>
+            <prepare>
+                <delete path="${eventsOutputPath}"/>
+            </prepare>
+            <main-class>eu.dnetlib.dhp.broker.oa.GenerateEventsApplication</main-class>
+            <arg>--graphPath</arg><arg>${graphInputPath}</arg>
+            <arg>--eventsPath</arg><arg>${eventsOutputPath}</arg>
+            <arg>--isLookupUrl</arg><arg>${isLookupUrl}</arg>
+            <arg>--dedupConfProfile</arg><arg>${dedupConfProfId}</arg>
+        </java>
+        <ok to="End"/>
+        <error to="Kill"/>
+    </action>
+
+    <end name="End"/>
+
+</workflow-app>

dhp-workflows/dhp-broker-events/src/main/resources/eu/dnetlib/dhp/broker/oa/generate_broker_events.json

@@ -0,0 +1,26 @@
+[
+	{
+		"paramName": "g",
+		"paramLongName": "graphPath",
+		"paramDescription": "the path where there the graph is stored",
+		"paramRequired": true
+	},
+	{
+		"paramName": "o",
+		"paramLongName": "eventsPath",
+		"paramDescription": "the path where the generated events will be stored",
+		"paramRequired": true
+	},
+	{
+		"paramName": "lu",
+		"paramLongName": "isLookupUrl",
+		"paramDescription": "the address of the ISLookUpService",
+		"paramRequired": true
+	},
+	{
+		"paramName": "d",
+		"paramLongName": "dedupConfProfile",
+		"paramDescription": "the id of a valid Dedup Configuration Profile",
+		"paramRequired": true
+	}
+]

dhp-workflows/dhp-dedup-openaire/pom.xml

@@ -3,7 +3,7 @@
     <parent>
         <artifactId>dhp-workflows</artifactId>
         <groupId>eu.dnetlib.dhp</groupId>
-        <version>1.2.2-SNAPSHOT</version>
+        <version>1.2.3-SNAPSHOT</version>
     </parent>
     <modelVersion>4.0.0</modelVersion>
     <artifactId>dhp-dedup-openaire</artifactId>

dhp-workflows/dhp-dedup-scholexplorer/pom.xml

@@ -3,7 +3,7 @@
     <parent>
         <artifactId>dhp-workflows</artifactId>
         <groupId>eu.dnetlib.dhp</groupId>
-        <version>1.2.2-SNAPSHOT</version>
+        <version>1.2.3-SNAPSHOT</version>
     </parent>
     <modelVersion>4.0.0</modelVersion>
 

dhp-workflows/dhp-distcp/pom.xml

@@ -3,7 +3,7 @@
     <parent>
         <artifactId>dhp-workflows</artifactId>
         <groupId>eu.dnetlib.dhp</groupId>
-        <version>1.2.2-SNAPSHOT</version>
+        <version>1.2.3-SNAPSHOT</version>
     </parent>
     <modelVersion>4.0.0</modelVersion>
 

dhp-workflows/dhp-doiboost/pom.xml

@@ -1,10 +1,9 @@
 <?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/xsd/maven-4.0.0.xsd">
+<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/xsd/maven-4.0.0.xsd">
     <parent>
         <artifactId>dhp-workflows</artifactId>
         <groupId>eu.dnetlib.dhp</groupId>
-        <version>1.2.2-SNAPSHOT</version>
+        <version>1.2.3-SNAPSHOT</version>
     </parent>
     <modelVersion>4.0.0</modelVersion>
 

dhp-workflows/dhp-doiboost/src/main/java/eu/dnetlib/doiboost/crossref/Crossref2Oaf.scala

@@ -166,8 +166,10 @@ case object Crossref2Oaf {
 
     val has_review = (json \ "relation" \"has-review" \ "id")
 
-    if(has_review != JNothing)
-      instance.setRefereed(asField("peerReviewed"))
+    if(has_review != JNothing) {
+      instance.setRefereed(
+        createQualifier("0001", "peerReviewed", "dnet:review_levels", "dnet:review_levels"))
+    }
 
 
     instance.setAccessright(getRestrictedQualifier())

dhp-workflows/dhp-enrichment/pom.xml

@@ -3,7 +3,7 @@
     <parent>
         <artifactId>dhp-workflows</artifactId>
         <groupId>eu.dnetlib.dhp</groupId>
-        <version>1.2.2-SNAPSHOT</version>
+        <version>1.2.3-SNAPSHOT</version>
     </parent>
     <modelVersion>4.0.0</modelVersion>
 

dhp-workflows/dhp-graph-mapper/pom.xml

@@ -3,7 +3,7 @@
     <parent>
         <artifactId>dhp-workflows</artifactId>
         <groupId>eu.dnetlib.dhp</groupId>
-        <version>1.2.2-SNAPSHOT</version>
+        <version>1.2.3-SNAPSHOT</version>
     </parent>
     <modelVersion>4.0.0</modelVersion>
 

dhp-workflows/dhp-graph-mapper/src/main/java/eu/dnetlib/dhp/oa/graph/clean/CleanGraphSparkJob.java

@@ -0,0 +1,109 @@
+
+package eu.dnetlib.dhp.oa.graph.clean;
+
+import static eu.dnetlib.dhp.common.SparkSessionSupport.runWithSparkSession;
+
+import java.util.Optional;
+
+import org.apache.commons.io.IOUtils;
+import org.apache.spark.SparkConf;
+import org.apache.spark.api.java.function.MapFunction;
+import org.apache.spark.sql.Dataset;
+import org.apache.spark.sql.Encoders;
+import org.apache.spark.sql.SaveMode;
+import org.apache.spark.sql.SparkSession;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import com.fasterxml.jackson.databind.ObjectMapper;
+
+import eu.dnetlib.dhp.application.ArgumentApplicationParser;
+import eu.dnetlib.dhp.common.HdfsSupport;
+import eu.dnetlib.dhp.oa.graph.raw.common.VocabularyGroup;
+import eu.dnetlib.dhp.schema.oaf.*;
+import eu.dnetlib.dhp.utils.ISLookupClientFactory;
+import eu.dnetlib.enabling.is.lookup.rmi.ISLookUpService;
+
+public class CleanGraphSparkJob {
+
+	private static final Logger log = LoggerFactory.getLogger(CleanGraphSparkJob.class);
+
+	private static final ObjectMapper OBJECT_MAPPER = new ObjectMapper();
+
+	public static void main(String[] args) throws Exception {
+
+		String jsonConfiguration = IOUtils
+			.toString(
+				CleanGraphSparkJob.class
+					.getResourceAsStream(
+						"/eu/dnetlib/dhp/oa/graph/input_clean_graph_parameters.json"));
+		final ArgumentApplicationParser parser = new ArgumentApplicationParser(jsonConfiguration);
+		parser.parseArgument(args);
+
+		Boolean isSparkSessionManaged = Optional
+			.ofNullable(parser.get("isSparkSessionManaged"))
+			.map(Boolean::valueOf)
+			.orElse(Boolean.TRUE);
+		log.info("isSparkSessionManaged: {}", isSparkSessionManaged);
+
+		String inputPath = parser.get("inputPath");
+		log.info("inputPath: {}", inputPath);
+
+		String outputPath = parser.get("outputPath");
+		log.info("outputPath: {}", outputPath);
+
+		String isLookupUrl = parser.get("isLookupUrl");
+		log.info("isLookupUrl: {}", isLookupUrl);
+
+		String graphTableClassName = parser.get("graphTableClassName");
+		log.info("graphTableClassName: {}", graphTableClassName);
+
+		Class<? extends OafEntity> entityClazz = (Class<? extends OafEntity>) Class.forName(graphTableClassName);
+
+		final ISLookUpService isLookupService = ISLookupClientFactory.getLookUpService(isLookupUrl);
+		final VocabularyGroup vocs = VocabularyGroup.loadVocsFromIS(isLookupService);
+
+		SparkConf conf = new SparkConf();
+		runWithSparkSession(
+			conf,
+			isSparkSessionManaged,
+			spark -> {
+				removeOutputDir(spark, outputPath);
+				fixGraphTable(spark, vocs, inputPath, entityClazz, outputPath);
+			});
+	}
+
+	private static <T extends Oaf> void fixGraphTable(
+		SparkSession spark,
+		VocabularyGroup vocs,
+		String inputPath,
+		Class<T> clazz,
+		String outputPath) {
+
+		final CleaningRuleMap mapping = CleaningRuleMap.create(vocs);
+
+		readTableFromPath(spark, inputPath, clazz)
+			.map((MapFunction<T, T>) value -> OafCleaner.apply(value, mapping), Encoders.bean(clazz))
+			.write()
+			.mode(SaveMode.Overwrite)
+			.option("compression", "gzip")
+			.json(outputPath);
+	}
+
+	private static <T extends Oaf> Dataset<T> readTableFromPath(
+		SparkSession spark, String inputEntityPath, Class<T> clazz) {
+
+		log.info("Reading Graph table from: {}", inputEntityPath);
+		return spark
+			.read()
+			.textFile(inputEntityPath)
+			.map(
+				(MapFunction<String, T>) value -> OBJECT_MAPPER.readValue(value, clazz),
+				Encoders.bean(clazz));
+	}
+
+	private static void removeOutputDir(SparkSession spark, String path) {
+		HdfsSupport.remove(path, spark.sparkContext().hadoopConfiguration());
+	}
+
+}

dhp-workflows/dhp-graph-mapper/src/main/java/eu/dnetlib/dhp/oa/graph/clean/CleaningRuleMap.java

@@ -0,0 +1,40 @@
+
+package eu.dnetlib.dhp.oa.graph.clean;
+
+import java.io.Serializable;
+import java.util.HashMap;
+
+import eu.dnetlib.dhp.common.FunctionalInterfaceSupport.SerializableConsumer;
+import eu.dnetlib.dhp.oa.graph.raw.common.VocabularyGroup;
+import eu.dnetlib.dhp.schema.oaf.Qualifier;
+import eu.dnetlib.dhp.schema.oaf.StructuredProperty;
+
+public class CleaningRuleMap extends HashMap<Class, SerializableConsumer<Object>> implements Serializable {
+
+	/**
+	 * Creates the mapping for the Oaf types subject to cleaning
+	 *
+	 * @param vocabularies
+	 */
+	public static CleaningRuleMap create(VocabularyGroup vocabularies) {
+		CleaningRuleMap mapping = new CleaningRuleMap();
+		mapping.put(Qualifier.class, o -> {
+			Qualifier q = (Qualifier) o;
+			if (vocabularies.vocabularyExists(q.getSchemeid())) {
+				Qualifier newValue = vocabularies.lookup(q.getSchemeid(), q.getClassid());
+				q.setClassid(newValue.getClassid());
+				q.setClassname(newValue.getClassname());
+			}
+		});
+		mapping.put(StructuredProperty.class, o -> {
+			StructuredProperty sp = (StructuredProperty) o;
+			// TODO implement a policy
+			/*
+			 * if (StringUtils.isBlank(sp.getValue())) { sp.setValue(null); sp.setQualifier(null); sp.setDataInfo(null);
+			 * }
+			 */
+		});
+		return mapping;
+	}
+
+}

dhp-workflows/dhp-graph-mapper/src/main/java/eu/dnetlib/dhp/oa/graph/clean/OafCleaner.java

@@ -0,0 +1,82 @@
+
+package eu.dnetlib.dhp.oa.graph.clean;
+
+import java.io.Serializable;
+import java.lang.reflect.Field;
+import java.util.Arrays;
+import java.util.LinkedList;
+import java.util.List;
+import java.util.Objects;
+
+import eu.dnetlib.dhp.schema.oaf.Oaf;
+
+public class OafCleaner implements Serializable {
+
+	public static <E extends Oaf> E apply(E oaf, CleaningRuleMap mapping) {
+		try {
+			navigate(oaf, mapping);
+		} catch (IllegalAccessException e) {
+			throw new RuntimeException(e);
+		}
+		return oaf;
+	}
+
+	private static void navigate(Object o, CleaningRuleMap mapping) throws IllegalAccessException {
+		if (isPrimitive(o)) {
+			return;
+		} else if (isIterable(o.getClass())) {
+			for (final Object elem : (Iterable<?>) o) {
+				navigate(elem, mapping);
+			}
+		} else if (hasMapping(o, mapping)) {
+			mapping.get(o.getClass()).accept(o);
+		} else {
+			for (final Field f : getAllFields(o.getClass())) {
+				f.setAccessible(true);
+				final Object val = f.get(o);
+				if (!isPrimitive(val) && hasMapping(val, mapping)) {
+					mapping.get(val.getClass()).accept(val);
+				} else {
+					navigate(f.get(o), mapping);
+				}
+			}
+		}
+	}
+
+	private static boolean hasMapping(Object o, CleaningRuleMap mapping) {
+		return mapping.containsKey(o.getClass());
+	}
+
+	private static boolean isIterable(final Class<?> cl) {
+		return Iterable.class.isAssignableFrom(cl);
+	}
+
+	private static boolean isPrimitive(Object o) {
+		return Objects.isNull(o)
+			|| o.getClass().isPrimitive()
+			|| o instanceof Class
+			|| o instanceof Integer
+			|| o instanceof Double
+			|| o instanceof Float
+			|| o instanceof Long
+			|| o instanceof Boolean
+			|| o instanceof String
+			|| o instanceof Byte;
+	}
+
+	private static List<Field> getAllFields(Class<?> clazz) {
+		return getAllFields(new LinkedList<>(), clazz);
+	}
+
+	private static List<Field> getAllFields(List<Field> fields, Class<?> clazz) {
+		fields.addAll(Arrays.asList(clazz.getDeclaredFields()));
+
+		final Class<?> superclass = clazz.getSuperclass();
+		if (Objects.nonNull(superclass) && superclass.getPackage().equals(Oaf.class.getPackage())) {
+			getAllFields(fields, superclass);
+		}
+
+		return fields;
+	}
+
+}

dhp-workflows/dhp-graph-mapper/src/main/java/eu/dnetlib/dhp/oa/graph/raw/GenerateEntitiesApplication.java

@@ -39,6 +39,8 @@ import eu.dnetlib.dhp.schema.oaf.Project;
 import eu.dnetlib.dhp.schema.oaf.Publication;
 import eu.dnetlib.dhp.schema.oaf.Relation;
 import eu.dnetlib.dhp.schema.oaf.Software;
+import eu.dnetlib.dhp.utils.ISLookupClientFactory;
+import eu.dnetlib.enabling.is.lookup.rmi.ISLookUpService;
 import scala.Tuple2;
 
 public class GenerateEntitiesApplication {
@@ -71,7 +73,8 @@ public class GenerateEntitiesApplication {
 		final String isLookupUrl = parser.get("isLookupUrl");
 		log.info("isLookupUrl: {}", isLookupUrl);
 
-		final VocabularyGroup vocs = VocabularyGroup.loadVocsFromIS(isLookupUrl);
+		final ISLookUpService isLookupService = ISLookupClientFactory.getLookUpService(isLookupUrl);
+		final VocabularyGroup vocs = VocabularyGroup.loadVocsFromIS(isLookupService);
 
 		final SparkConf conf = new SparkConf();
 		runWithSparkSession(conf, isSparkSessionManaged, spark -> {
@@ -137,11 +140,11 @@ public class GenerateEntitiesApplication {
 		final String type = StringUtils.substringAfter(id, ":");
 
 		switch (type.toLowerCase()) {
-			case "oaf-store-claim":
 			case "oaf-store-cleaned":
+			case "oaf-store-claim":
 				return new OafToOafMapper(vocs, false).processMdRecord(s);
-			case "odf-store-claim":
 			case "odf-store-cleaned":
+			case "odf-store-claim":
 				return new OdfToOafMapper(vocs, false).processMdRecord(s);
 			case "oaf-store-intersection":
 				return new OafToOafMapper(vocs, true).processMdRecord(s);

dhp-workflows/dhp-graph-mapper/src/main/java/eu/dnetlib/dhp/oa/graph/raw/MigrateDbEntitiesApplication.java

@@ -50,6 +50,8 @@ import org.apache.commons.lang3.StringUtils;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
+import com.google.common.collect.Lists;
+
 import eu.dnetlib.dhp.application.ArgumentApplicationParser;
 import eu.dnetlib.dhp.common.DbClient;
 import eu.dnetlib.dhp.oa.graph.raw.common.AbstractMigrationApplication;
@@ -71,6 +73,7 @@ import eu.dnetlib.dhp.schema.oaf.Relation;
 import eu.dnetlib.dhp.schema.oaf.Result;
 import eu.dnetlib.dhp.schema.oaf.Software;
 import eu.dnetlib.dhp.schema.oaf.StructuredProperty;
+import eu.dnetlib.dhp.utils.ISLookupClientFactory;
 
 public class MigrateDbEntitiesApplication extends AbstractMigrationApplication implements Closeable {
 
@@ -151,7 +154,7 @@ public class MigrateDbEntitiesApplication extends AbstractMigrationApplication i
 		super(hdfsPath);
 		this.dbClient = new DbClient(dbUrl, dbUser, dbPassword);
 		this.lastUpdateTimestamp = new Date().getTime();
-		this.vocs = VocabularyGroup.loadVocsFromIS(isLookupUrl);
+		this.vocs = VocabularyGroup.loadVocsFromIS(ISLookupClientFactory.getLookUpService(isLookupUrl));
 	}
 
 	public void execute(final String sqlFile, final Function<ResultSet, List<Oaf>> producer)
@@ -170,7 +173,7 @@ public class MigrateDbEntitiesApplication extends AbstractMigrationApplication i
 			final Datasource ds = new Datasource();
 
 			ds.setId(createOpenaireId(10, rs.getString("datasourceid"), true));
-			ds.setOriginalId(Arrays.asList(rs.getString("datasourceid")));
+			ds.setOriginalId(Arrays.asList((String[]) rs.getArray("identities").getArray()));
 			ds
 				.setCollectedfrom(
 					listKeyValues(

dhp-workflows/dhp-graph-mapper/src/main/java/eu/dnetlib/dhp/oa/graph/raw/OafToOafMapper.java

@@ -4,13 +4,7 @@ package eu.dnetlib.dhp.oa.graph.raw;
 import static eu.dnetlib.dhp.oa.graph.raw.common.OafMapperUtils.createOpenaireId;
 import static eu.dnetlib.dhp.oa.graph.raw.common.OafMapperUtils.field;
 import static eu.dnetlib.dhp.oa.graph.raw.common.OafMapperUtils.structuredProperty;
-import static eu.dnetlib.dhp.schema.common.ModelConstants.DNET_ACCESS_MODES;
-import static eu.dnetlib.dhp.schema.common.ModelConstants.DNET_LANGUAGES;
-import static eu.dnetlib.dhp.schema.common.ModelConstants.DNET_PID_TYPES;
-import static eu.dnetlib.dhp.schema.common.ModelConstants.DNET_PUBLICATION_RESOURCE;
-import static eu.dnetlib.dhp.schema.common.ModelConstants.IS_RELATED_TO;
-import static eu.dnetlib.dhp.schema.common.ModelConstants.PUBLICATION_DATASET;
-import static eu.dnetlib.dhp.schema.common.ModelConstants.RESULT_RESULT;
+import static eu.dnetlib.dhp.schema.common.ModelConstants.*;
 
 import java.util.ArrayList;
 import java.util.List;
@@ -139,7 +133,7 @@ public class OafToOafMapper extends AbstractMdRecordToOafMapper {
 		instance
 			.setAccessright(prepareQualifier(doc, "//oaf:accessrights", DNET_ACCESS_MODES));
 		instance.setLicense(field(doc.valueOf("//oaf:license"), info));
-		instance.setRefereed(field(doc.valueOf("//oaf:refereed"), info));
+		instance.setRefereed(prepareQualifier(doc, "//oaf:refereed", DNET_REVIEW_LEVELS));
 		instance
 			.setProcessingchargeamount(field(doc.valueOf("//oaf:processingchargeamount"), info));
 		instance
@@ -281,12 +275,12 @@ public class OafToOafMapper extends AbstractMdRecordToOafMapper {
 				res
 					.add(
 						getRelation(
-							docId, otherId, RESULT_RESULT, PUBLICATION_DATASET, IS_RELATED_TO, collectedFrom, info,
+							docId, otherId, RESULT_RESULT, RELATIONSHIP, IS_RELATED_TO, collectedFrom, info,
 							lastUpdateTimestamp));
 				res
 					.add(
 						getRelation(
-							otherId, docId, RESULT_RESULT, PUBLICATION_DATASET, IS_RELATED_TO, collectedFrom, info,
+							otherId, docId, RESULT_RESULT, RELATIONSHIP, IS_RELATED_TO, collectedFrom, info,
 							lastUpdateTimestamp));
 			}
 		}

dhp-workflows/dhp-graph-mapper/src/main/java/eu/dnetlib/dhp/oa/graph/raw/OdfToOafMapper.java

@@ -4,19 +4,7 @@ package eu.dnetlib.dhp.oa.graph.raw;
 import static eu.dnetlib.dhp.oa.graph.raw.common.OafMapperUtils.createOpenaireId;
 import static eu.dnetlib.dhp.oa.graph.raw.common.OafMapperUtils.field;
 import static eu.dnetlib.dhp.oa.graph.raw.common.OafMapperUtils.structuredProperty;
-import static eu.dnetlib.dhp.schema.common.ModelConstants.DNET_ACCESS_MODES;
-import static eu.dnetlib.dhp.schema.common.ModelConstants.DNET_DATA_CITE_DATE;
-import static eu.dnetlib.dhp.schema.common.ModelConstants.DNET_DATA_CITE_RESOURCE;
-import static eu.dnetlib.dhp.schema.common.ModelConstants.DNET_LANGUAGES;
-import static eu.dnetlib.dhp.schema.common.ModelConstants.DNET_PID_TYPES;
-import static eu.dnetlib.dhp.schema.common.ModelConstants.DNET_PUBLICATION_RESOURCE;
-import static eu.dnetlib.dhp.schema.common.ModelConstants.HAS_PARTS;
-import static eu.dnetlib.dhp.schema.common.ModelConstants.IS_PART_OF;
-import static eu.dnetlib.dhp.schema.common.ModelConstants.IS_SUPPLEMENTED_BY;
-import static eu.dnetlib.dhp.schema.common.ModelConstants.IS_SUPPLEMENT_TO;
-import static eu.dnetlib.dhp.schema.common.ModelConstants.PART;
-import static eu.dnetlib.dhp.schema.common.ModelConstants.RESULT_RESULT;
-import static eu.dnetlib.dhp.schema.common.ModelConstants.SUPPLEMENT;
+import static eu.dnetlib.dhp.schema.common.ModelConstants.*;
 
 import java.util.ArrayList;
 import java.util.Arrays;
@@ -129,7 +117,7 @@ public class OdfToOafMapper extends AbstractMdRecordToOafMapper {
 		instance
 			.setAccessright(prepareQualifier(doc, "//oaf:accessrights", DNET_ACCESS_MODES));
 		instance.setLicense(field(doc.valueOf("//oaf:license"), info));
-		instance.setRefereed(field(doc.valueOf("//oaf:refereed"), info));
+		instance.setRefereed(prepareQualifier(doc, "//oaf:refereed", DNET_REVIEW_LEVELS));
 		instance.setProcessingchargeamount(field(doc.valueOf("//oaf:processingchargeamount"), info));
 		instance
 			.setProcessingchargecurrency(field(doc.valueOf("//oaf:processingchargeamount/@currency"), info));

dhp-workflows/dhp-graph-mapper/src/main/java/eu/dnetlib/dhp/oa/graph/raw/common/OafMapperUtils.java

@@ -60,6 +60,10 @@ public class OafMapperUtils {
 			.collect(Collectors.toList());
 	}
 
+	public static Qualifier unknown(final String schemeid, final String schemename) {
+		return qualifier("UNKNOWN", "Unknown", schemeid, schemename);
+	}
+
 	public static Qualifier qualifier(
 		final String classid,
 		final String classname,

dhp-workflows/dhp-graph-mapper/src/main/java/eu/dnetlib/dhp/oa/graph/raw/common/Vocabulary.java

@@ -4,14 +4,29 @@ package eu.dnetlib.dhp.oa.graph.raw.common;
 import java.io.Serializable;
 import java.util.HashMap;
 import java.util.Map;
+import java.util.Optional;
+
+import org.apache.commons.lang3.StringUtils;
+
+import com.google.common.collect.Maps;
+
+import eu.dnetlib.dhp.schema.oaf.Qualifier;
 
 public class Vocabulary implements Serializable {
 
 	private final String id;
 	private final String name;
 
+	/**
+	 * Code to Term mappings for this Vocabulary.
+	 */
 	private final Map<String, VocabularyTerm> terms = new HashMap<>();
 
+	/**
+	 * Synonym to Code mappings for this Vocabulary.
+	 */
+	private final Map<String, String> synonyms = Maps.newHashMap();
+
 	public Vocabulary(final String id, final String name) {
 		this.id = id;
 		this.name = name;
@@ -30,7 +45,7 @@ public class Vocabulary implements Serializable {
 	}
 
 	public VocabularyTerm getTerm(final String id) {
-		return terms.get(id.toLowerCase());
+		return Optional.ofNullable(id).map(s -> s.toLowerCase()).map(s -> terms.get(s)).orElse(null);
 	}
 
 	protected void addTerm(final String id, final String name) {
@@ -40,4 +55,32 @@ public class Vocabulary implements Serializable {
 	protected boolean termExists(final String id) {
 		return terms.containsKey(id.toLowerCase());
 	}
+
+	protected void addSynonym(final String syn, final String termCode) {
+		synonyms.put(syn, termCode.toLowerCase());
+	}
+
+	public VocabularyTerm getTermBySynonym(final String syn) {
+		return getTerm(synonyms.get(syn));
+	}
+
+	public Qualifier getTermAsQualifier(final String termId) {
+		if (StringUtils.isBlank(termId)) {
+			return OafMapperUtils.unknown(getId(), getName());
+		} else if (termExists(termId)) {
+			final VocabularyTerm t = getTerm(termId);
+			return OafMapperUtils.qualifier(t.getId(), t.getName(), getId(), getName());
+		} else {
+			return OafMapperUtils.qualifier(termId, termId, getId(), getName());
+		}
+	}
+
+	public Qualifier getSynonymAsQualifier(final String syn) {
+		return Optional
+			.ofNullable(getTermBySynonym(syn))
+			.map(term -> getTermAsQualifier(term.getId()))
+			.orElse(null);
+		// .orElse(OafMapperUtils.unknown(getId(), getName()));
+	}
+
 }

dhp-workflows/dhp-graph-mapper/src/main/java/eu/dnetlib/dhp/oa/graph/raw/common/VocabularyGroup.java

@@ -1,33 +1,39 @@
 
 package eu.dnetlib.dhp.oa.graph.raw.common;
 
-import java.io.IOException;
 import java.io.Serializable;
-import java.util.HashMap;
-import java.util.Map;
+import java.util.*;
+import java.util.function.Supplier;
+import java.util.stream.Collectors;
 
-import org.apache.commons.io.IOUtils;
 import org.apache.commons.lang3.StringUtils;
 
-import eu.dnetlib.dhp.oa.graph.raw.GenerateEntitiesApplication;
 import eu.dnetlib.dhp.schema.oaf.Qualifier;
-import eu.dnetlib.dhp.utils.ISLookupClientFactory;
 import eu.dnetlib.enabling.is.lookup.rmi.ISLookUpException;
 import eu.dnetlib.enabling.is.lookup.rmi.ISLookUpService;
 
 public class VocabularyGroup implements Serializable {
 
-	public static VocabularyGroup loadVocsFromIS(final String isLookupUrl) throws IOException, ISLookUpException {
-		final ISLookUpService isLookUpService = ISLookupClientFactory.getLookUpService(isLookupUrl);
+	public static final String VOCABULARIES_XQUERY = "for $x in collection(' /db/DRIVER/VocabularyDSResources/VocabularyDSResourceType') \n"
+		+
+		"let $vocid := $x//VOCABULARY_NAME/@code\n" +
+		"let $vocname := $x//VOCABULARY_NAME/text()\n" +
+		"for $term in ($x//TERM)\n" +
+		"return concat($vocid,' @=@ ',$vocname,' @=@ ',$term/@code,' @=@ ',$term/@english_name)";
 
-		final String xquery = IOUtils
-			.toString(
-				GenerateEntitiesApplication.class
-					.getResourceAsStream("/eu/dnetlib/dhp/oa/graph/xquery/load_vocabularies.xquery"));
+	public static final String VOCABULARY_SYNONYMS_XQUERY = "for $x in collection('/db/DRIVER/VocabularyDSResources/VocabularyDSResourceType')\n"
+		+
+		"let $vocid := $x//VOCABULARY_NAME/@code\n" +
+		"let $vocname := $x//VOCABULARY_NAME/text()\n" +
+		"for $term in ($x//TERM)\n" +
+		"for $syn in ($term//SYNONYM/@term)\n" +
+		"return concat($vocid,' @=@ ',$term/@code,' @=@ ', $syn)\n";
+
+	public static VocabularyGroup loadVocsFromIS(ISLookUpService isLookUpService) throws ISLookUpException {
 
 		final VocabularyGroup vocs = new VocabularyGroup();
 
-		for (final String s : isLookUpService.quickSearchProfile(xquery)) {
+		for (final String s : isLookUpService.quickSearchProfile(VOCABULARIES_XQUERY)) {
 			final String[] arr = s.split("@=@");
 			if (arr.length == 4) {
 				final String vocId = arr[0].trim();
@@ -40,6 +46,19 @@ public class VocabularyGroup implements Serializable {
 				}
 
 				vocs.addTerm(vocId, termId, termName);
+				vocs.addSynonyms(vocId, termId, termId);
+			}
+		}
+
+		for (final String s : isLookUpService.quickSearchProfile(VOCABULARY_SYNONYMS_XQUERY)) {
+			final String[] arr = s.split("@=@");
+			if (arr.length == 3) {
+				final String vocId = arr[0].trim();
+				final String termId = arr[1].trim();
+				final String syn = arr[2].trim();
+
+				vocs.addSynonyms(vocId, termId, syn);
+				vocs.addSynonyms(vocId, termId, termId);
 			}
 		}
 
@@ -66,16 +85,37 @@ public class VocabularyGroup implements Serializable {
 		}
 	}
 
-	public Qualifier getTermAsQualifier(final String vocId, final String id) {
-		if (StringUtils.isBlank(id)) {
-			return OafMapperUtils.qualifier("UNKNOWN", "UNKNOWN", vocId, vocId);
-		} else if (termExists(vocId, id)) {
-			final Vocabulary v = vocs.get(vocId.toLowerCase());
-			final VocabularyTerm t = v.getTerm(id);
-			return OafMapperUtils.qualifier(t.getId(), t.getName(), v.getId(), v.getName());
-		} else {
-			return OafMapperUtils.qualifier(id, id, vocId, vocId);
+	public Set<String> getTerms(String vocId) {
+		if (!vocabularyExists(vocId)) {
+			return new HashSet<>();
 		}
+		return vocs
+			.get(vocId.toLowerCase())
+			.getTerms()
+			.values()
+			.stream()
+			.map(t -> t.getId())
+			.collect(Collectors.toCollection(HashSet::new));
+	}
+
+	public Qualifier lookup(String vocId, String id) {
+		return Optional
+			.ofNullable(getSynonymAsQualifier(vocId, id))
+			.orElse(getTermAsQualifier(vocId, id));
+	}
+
+	public Qualifier getTermAsQualifier(final String vocId, final String id) {
+		if (vocabularyExists(vocId)) {
+			return vocs.get(vocId.toLowerCase()).getTermAsQualifier(id);
+		}
+		return OafMapperUtils.qualifier(id, id, "", "");
+	}
+
+	public Qualifier getSynonymAsQualifier(final String vocId, final String syn) {
+		if (StringUtils.isBlank(vocId)) {
+			return OafMapperUtils.unknown("", "");
+		}
+		return vocs.get(vocId.toLowerCase()).getSynonymAsQualifier(syn);
 	}
 
 	public boolean termExists(final String vocId, final String id) {
@@ -86,4 +126,16 @@ public class VocabularyGroup implements Serializable {
 		return vocs.containsKey(vocId.toLowerCase());
 	}
 
+	private void addSynonyms(final String vocId, final String termId, final String syn) {
+		String id = Optional
+			.ofNullable(vocId)
+			.map(s -> s.toLowerCase())
+			.orElseThrow(
+				() -> new IllegalArgumentException(String.format("empty vocabulary id for [term:%s, synonym:%s]")));
+		Optional
+			.ofNullable(vocs.get(id))
+			.orElseThrow(() -> new IllegalArgumentException("missing vocabulary id: " + vocId))
+			.addSynonym(syn, termId);
+	}
+
 }

dhp-workflows/dhp-graph-mapper/src/main/resources/eu/dnetlib/dhp/oa/graph/clean/oozie_app/config-default.xml

@@ -0,0 +1,18 @@
+<configuration>
+    <property>
+        <name>jobTracker</name>
+        <value>yarnRM</value>
+    </property>
+    <property>
+        <name>nameNode</name>
+        <value>hdfs://nameservice1</value>
+    </property>
+    <property>
+        <name>oozie.use.system.libpath</name>
+        <value>true</value>
+    </property>
+    <property>
+        <name>oozie.action.sharelib.for.spark</name>
+        <value>spark2</value>
+    </property>
+</configuration>

dhp-workflows/dhp-graph-mapper/src/main/resources/eu/dnetlib/dhp/oa/graph/clean/oozie_app/workflow.xml

@@ -0,0 +1,281 @@
+<workflow-app name="clean graph" xmlns="uri:oozie:workflow:0.5">
+
+    <parameters>
+        <property>
+            <name>graphInputPath</name>
+            <description>the input path to read graph content</description>
+        </property>
+        <property>
+            <name>graphOutputPath</name>
+            <description>the target path to store cleaned graph</description>
+        </property>
+        <property>
+            <name>isLookupUrl</name>
+            <description>the address of the lookUp service</description>
+        </property>
+
+        <property>
+            <name>sparkDriverMemory</name>
+            <description>memory for driver process</description>
+        </property>
+        <property>
+            <name>sparkExecutorMemory</name>
+            <description>memory for individual executor</description>
+        </property>
+        <property>
+            <name>sparkExecutorCores</name>
+            <description>number of cores used by single executor</description>
+        </property>
+        <property>
+            <name>oozieActionShareLibForSpark2</name>
+            <description>oozie action sharelib for spark 2.*</description>
+        </property>
+        <property>
+            <name>spark2ExtraListeners</name>
+            <value>com.cloudera.spark.lineage.NavigatorAppListener</value>
+            <description>spark 2.* extra listeners classname</description>
+        </property>
+        <property>
+            <name>spark2SqlQueryExecutionListeners</name>
+            <value>com.cloudera.spark.lineage.NavigatorQueryListener</value>
+            <description>spark 2.* sql query execution listeners classname</description>
+        </property>
+        <property>
+            <name>spark2YarnHistoryServerAddress</name>
+            <description>spark 2.* yarn history server address</description>
+        </property>
+        <property>
+            <name>spark2EventLogDir</name>
+            <description>spark 2.* event log dir location</description>
+        </property>
+    </parameters>
+
+	<start to="fork_clean_graph"/>
+
+    <kill name="Kill">
+        <message>Action failed, error message[${wf:errorMessage(wf:lastErrorNode())}]</message>
+    </kill>
+
+    <fork name="fork_clean_graph">
+        <path start="clean_publication"/>
+        <path start="clean_dataset"/>
+        <path start="clean_otherresearchproduct"/>
+        <path start="clean_software"/>
+        <path start="clean_datasource"/>
+        <path start="clean_organization"/>
+        <path start="clean_project"/>
+        <path start="clean_relation"/>
+    </fork>
+
+    <action name="clean_publication">
+        <spark xmlns="uri:oozie:spark-action:0.2">
+            <master>yarn</master>
+            <mode>cluster</mode>
+            <name>Clean publications</name>
+            <class>eu.dnetlib.dhp.oa.graph.clean.CleanGraphSparkJob</class>
+            <jar>dhp-graph-mapper-${projectVersion}.jar</jar>
+            <spark-opts>
+                --executor-cores=${sparkExecutorCores}
+                --executor-memory=${sparkExecutorMemory}
+                --driver-memory=${sparkDriverMemory}
+                --conf spark.extraListeners=${spark2ExtraListeners}
+                --conf spark.sql.queryExecutionListeners=${spark2SqlQueryExecutionListeners}
+                --conf spark.yarn.historyServer.address=${spark2YarnHistoryServerAddress}
+                --conf spark.eventLog.dir=${nameNode}${spark2EventLogDir}
+                --conf spark.sql.shuffle.partitions=7680
+            </spark-opts>
+            <arg>--inputPath</arg><arg>${graphInputPath}/publication</arg>
+            <arg>--outputPath</arg><arg>${graphOutputPath}/publication</arg>
+            <arg>--graphTableClassName</arg><arg>eu.dnetlib.dhp.schema.oaf.Publication</arg>
+            <arg>--isLookupUrl</arg><arg>${isLookupUrl}</arg>
+        </spark>
+        <ok to="wait_clean"/>
+        <error to="Kill"/>
+    </action>
+
+    <action name="clean_dataset">
+        <spark xmlns="uri:oozie:spark-action:0.2">
+            <master>yarn</master>
+            <mode>cluster</mode>
+            <name>Clean datasets</name>
+            <class>eu.dnetlib.dhp.oa.graph.clean.CleanGraphSparkJob</class>
+            <jar>dhp-graph-mapper-${projectVersion}.jar</jar>
+            <spark-opts>
+                --executor-cores=${sparkExecutorCores}
+                --executor-memory=${sparkExecutorMemory}
+                --driver-memory=${sparkDriverMemory}
+                --conf spark.extraListeners=${spark2ExtraListeners}
+                --conf spark.sql.queryExecutionListeners=${spark2SqlQueryExecutionListeners}
+                --conf spark.yarn.historyServer.address=${spark2YarnHistoryServerAddress}
+                --conf spark.eventLog.dir=${nameNode}${spark2EventLogDir}
+                --conf spark.sql.shuffle.partitions=7680
+            </spark-opts>
+            <arg>--inputPath</arg><arg>${graphInputPath}/dataset</arg>
+            <arg>--outputPath</arg><arg>${graphOutputPath}/dataset</arg>
+            <arg>--graphTableClassName</arg><arg>eu.dnetlib.dhp.schema.oaf.Dataset</arg>
+            <arg>--isLookupUrl</arg><arg>${isLookupUrl}</arg>
+        </spark>
+        <ok to="wait_clean"/>
+        <error to="Kill"/>
+    </action>
+
+    <action name="clean_otherresearchproduct">
+        <spark xmlns="uri:oozie:spark-action:0.2">
+            <master>yarn</master>
+            <mode>cluster</mode>
+            <name>Clean otherresearchproducts</name>
+            <class>eu.dnetlib.dhp.oa.graph.clean.CleanGraphSparkJob</class>
+            <jar>dhp-graph-mapper-${projectVersion}.jar</jar>
+            <spark-opts>
+                --executor-cores=${sparkExecutorCores}
+                --executor-memory=${sparkExecutorMemory}
+                --driver-memory=${sparkDriverMemory}
+                --conf spark.extraListeners=${spark2ExtraListeners}
+                --conf spark.sql.queryExecutionListeners=${spark2SqlQueryExecutionListeners}
+                --conf spark.yarn.historyServer.address=${spark2YarnHistoryServerAddress}
+                --conf spark.eventLog.dir=${nameNode}${spark2EventLogDir}
+                --conf spark.sql.shuffle.partitions=7680
+            </spark-opts>
+            <arg>--inputPath</arg><arg>${graphInputPath}/otherresearchproduct</arg>
+            <arg>--outputPath</arg><arg>${graphOutputPath}/otherresearchproduct</arg>
+            <arg>--graphTableClassName</arg><arg>eu.dnetlib.dhp.schema.oaf.OtherResearchProduct</arg>
+            <arg>--isLookupUrl</arg><arg>${isLookupUrl}</arg>
+        </spark>
+        <ok to="wait_clean"/>
+        <error to="Kill"/>
+    </action>
+
+    <action name="clean_software">
+        <spark xmlns="uri:oozie:spark-action:0.2">
+            <master>yarn</master>
+            <mode>cluster</mode>
+            <name>Clean softwares</name>
+            <class>eu.dnetlib.dhp.oa.graph.clean.CleanGraphSparkJob</class>
+            <jar>dhp-graph-mapper-${projectVersion}.jar</jar>
+            <spark-opts>
+                --executor-cores=${sparkExecutorCores}
+                --executor-memory=${sparkExecutorMemory}
+                --driver-memory=${sparkDriverMemory}
+                --conf spark.extraListeners=${spark2ExtraListeners}
+                --conf spark.sql.queryExecutionListeners=${spark2SqlQueryExecutionListeners}
+                --conf spark.yarn.historyServer.address=${spark2YarnHistoryServerAddress}
+                --conf spark.eventLog.dir=${nameNode}${spark2EventLogDir}
+                --conf spark.sql.shuffle.partitions=7680
+            </spark-opts>
+            <arg>--inputPath</arg><arg>${graphInputPath}/software</arg>
+            <arg>--outputPath</arg><arg>${graphOutputPath}/software</arg>
+            <arg>--graphTableClassName</arg><arg>eu.dnetlib.dhp.schema.oaf.Software</arg>
+            <arg>--isLookupUrl</arg><arg>${isLookupUrl}</arg>
+        </spark>
+        <ok to="wait_clean"/>
+        <error to="Kill"/>
+    </action>
+
+    <action name="clean_datasource">
+        <spark xmlns="uri:oozie:spark-action:0.2">
+            <master>yarn</master>
+            <mode>cluster</mode>
+            <name>Clean datasources</name>
+            <class>eu.dnetlib.dhp.oa.graph.clean.CleanGraphSparkJob</class>
+            <jar>dhp-graph-mapper-${projectVersion}.jar</jar>
+            <spark-opts>
+                --executor-cores=${sparkExecutorCores}
+                --executor-memory=${sparkExecutorMemory}
+                --driver-memory=${sparkDriverMemory}
+                --conf spark.extraListeners=${spark2ExtraListeners}
+                --conf spark.sql.queryExecutionListeners=${spark2SqlQueryExecutionListeners}
+                --conf spark.yarn.historyServer.address=${spark2YarnHistoryServerAddress}
+                --conf spark.eventLog.dir=${nameNode}${spark2EventLogDir}
+                --conf spark.sql.shuffle.partitions=7680
+            </spark-opts>
+            <arg>--inputPath</arg><arg>${graphInputPath}/datasource</arg>
+            <arg>--outputPath</arg><arg>${graphOutputPath}/datasource</arg>
+            <arg>--graphTableClassName</arg><arg>eu.dnetlib.dhp.schema.oaf.Datasource</arg>
+            <arg>--isLookupUrl</arg><arg>${isLookupUrl}</arg>
+        </spark>
+        <ok to="wait_clean"/>
+        <error to="Kill"/>
+    </action>
+
+    <action name="clean_organization">
+        <spark xmlns="uri:oozie:spark-action:0.2">
+            <master>yarn</master>
+            <mode>cluster</mode>
+            <name>Clean organizations</name>
+            <class>eu.dnetlib.dhp.oa.graph.clean.CleanGraphSparkJob</class>
+            <jar>dhp-graph-mapper-${projectVersion}.jar</jar>
+            <spark-opts>
+                --executor-cores=${sparkExecutorCores}
+                --executor-memory=${sparkExecutorMemory}
+                --driver-memory=${sparkDriverMemory}
+                --conf spark.extraListeners=${spark2ExtraListeners}
+                --conf spark.sql.queryExecutionListeners=${spark2SqlQueryExecutionListeners}
+                --conf spark.yarn.historyServer.address=${spark2YarnHistoryServerAddress}
+                --conf spark.eventLog.dir=${nameNode}${spark2EventLogDir}
+                --conf spark.sql.shuffle.partitions=7680
+            </spark-opts>
+            <arg>--inputPath</arg><arg>${graphInputPath}/organization</arg>
+            <arg>--outputPath</arg><arg>${graphOutputPath}/organization</arg>
+            <arg>--graphTableClassName</arg><arg>eu.dnetlib.dhp.schema.oaf.Organization</arg>
+            <arg>--isLookupUrl</arg><arg>${isLookupUrl}</arg>
+        </spark>
+        <ok to="wait_clean"/>
+        <error to="Kill"/>
+    </action>
+
+    <action name="clean_project">
+        <spark xmlns="uri:oozie:spark-action:0.2">
+            <master>yarn</master>
+            <mode>cluster</mode>
+            <name>Clean projects</name>
+            <class>eu.dnetlib.dhp.oa.graph.clean.CleanGraphSparkJob</class>
+            <jar>dhp-graph-mapper-${projectVersion}.jar</jar>
+            <spark-opts>
+                --executor-cores=${sparkExecutorCores}
+                --executor-memory=${sparkExecutorMemory}
+                --driver-memory=${sparkDriverMemory}
+                --conf spark.extraListeners=${spark2ExtraListeners}
+                --conf spark.sql.queryExecutionListeners=${spark2SqlQueryExecutionListeners}
+                --conf spark.yarn.historyServer.address=${spark2YarnHistoryServerAddress}
+                --conf spark.eventLog.dir=${nameNode}${spark2EventLogDir}
+                --conf spark.sql.shuffle.partitions=7680
+            </spark-opts>
+            <arg>--inputPath</arg><arg>${graphInputPath}/project</arg>
+            <arg>--outputPath</arg><arg>${graphOutputPath}/project</arg>
+            <arg>--graphTableClassName</arg><arg>eu.dnetlib.dhp.schema.oaf.Project</arg>
+            <arg>--isLookupUrl</arg><arg>${isLookupUrl}</arg>
+        </spark>
+        <ok to="wait_clean"/>
+        <error to="Kill"/>
+    </action>
+
+    <action name="clean_relation">
+        <spark xmlns="uri:oozie:spark-action:0.2">
+            <master>yarn</master>
+            <mode>cluster</mode>
+            <name>Clean relations</name>
+            <class>eu.dnetlib.dhp.oa.graph.clean.CleanGraphSparkJob</class>
+            <jar>dhp-graph-mapper-${projectVersion}.jar</jar>
+            <spark-opts>
+                --executor-cores=${sparkExecutorCores}
+                --executor-memory=${sparkExecutorMemory}
+                --driver-memory=${sparkDriverMemory}
+                --conf spark.extraListeners=${spark2ExtraListeners}
+                --conf spark.sql.queryExecutionListeners=${spark2SqlQueryExecutionListeners}
+                --conf spark.yarn.historyServer.address=${spark2YarnHistoryServerAddress}
+                --conf spark.eventLog.dir=${nameNode}${spark2EventLogDir}
+                --conf spark.sql.shuffle.partitions=7680
+            </spark-opts>
+            <arg>--inputPath</arg><arg>${graphInputPath}/relation</arg>
+            <arg>--outputPath</arg><arg>${graphOutputPath}/relation</arg>
+            <arg>--graphTableClassName</arg><arg>eu.dnetlib.dhp.schema.oaf.Relation</arg>
+            <arg>--isLookupUrl</arg><arg>${isLookupUrl}</arg>
+        </spark>
+        <ok to="wait_clean"/>
+        <error to="Kill"/>
+    </action>
+
+    <join name="wait_clean" to="End"/>
+
+    <end name="End"/>
+</workflow-app>

dhp-workflows/dhp-graph-mapper/src/main/resources/eu/dnetlib/dhp/oa/graph/input_clean_graph_parameters.json

@@ -0,0 +1,32 @@
+[
+  {
+    "paramName": "issm",
+    "paramLongName": "isSparkSessionManaged",
+    "paramDescription": "when true will stop SparkSession after job execution",
+    "paramRequired": false
+  },
+  {
+    "paramName": "in",
+    "paramLongName": "inputPath",
+    "paramDescription": "the path to the graph data dump to read",
+    "paramRequired": true
+  },
+  {
+    "paramName": "out",
+    "paramLongName": "outputPath",
+    "paramDescription": "the path to store the output graph",
+    "paramRequired": true
+  },
+  {
+    "paramName": "isu",
+    "paramLongName": "isLookupUrl",
+    "paramDescription": "url to the ISLookup Service",
+    "paramRequired": true
+  },
+  {
+    "paramName": "class",
+    "paramLongName": "graphTableClassName",
+    "paramDescription": "class name moelling the graph table",
+    "paramRequired": true
+  }
+]

dhp-workflows/dhp-graph-mapper/src/main/resources/eu/dnetlib/dhp/oa/graph/xquery/load_vocabularies_synonyms.xquery

@@ -0,0 +1,6 @@
+for $x in collection('/db/DRIVER/VocabularyDSResources/VocabularyDSResourceType')
+let $vocid := $x//VOCABULARY_NAME/@code
+let $vocname := $x//VOCABULARY_NAME/text()
+	for $term in ($x//TERM)
+		for $syn in ($term//SYNONYM/@term)
+			return concat($vocid,' @=@ ',$term/@code,' @=@ ', $syn)

dhp-workflows/dhp-graph-mapper/src/test/java/eu/dnetlib/dhp/oa/graph/clean/CleaningFunctionTest.java

@@ -0,0 +1,106 @@
+
+package eu.dnetlib.dhp.oa.graph.clean;
+
+import static org.junit.jupiter.api.Assertions.*;
+import static org.mockito.Mockito.lenient;
+
+import java.io.IOException;
+import java.util.List;
+import java.util.Set;
+import java.util.stream.Stream;
+
+import org.apache.commons.io.IOUtils;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.ExtendWith;
+import org.mockito.Mock;
+import org.mockito.junit.jupiter.MockitoExtension;
+
+import com.fasterxml.jackson.databind.ObjectMapper;
+
+import eu.dnetlib.dhp.oa.graph.raw.common.VocabularyGroup;
+import eu.dnetlib.dhp.schema.oaf.Publication;
+import eu.dnetlib.dhp.schema.oaf.Qualifier;
+import eu.dnetlib.enabling.is.lookup.rmi.ISLookUpException;
+import eu.dnetlib.enabling.is.lookup.rmi.ISLookUpService;
+
+@ExtendWith(MockitoExtension.class)
+public class CleaningFunctionTest {
+
+	public static final ObjectMapper MAPPER = new ObjectMapper();
+
+	@Mock
+	private ISLookUpService isLookUpService;
+
+	private VocabularyGroup vocabularies;
+
+	private CleaningRuleMap mapping;
+
+	@BeforeEach
+	public void setUp() throws ISLookUpException, IOException {
+		lenient().when(isLookUpService.quickSearchProfile(VocabularyGroup.VOCABULARIES_XQUERY)).thenReturn(vocs());
+		lenient()
+			.when(isLookUpService.quickSearchProfile(VocabularyGroup.VOCABULARY_SYNONYMS_XQUERY))
+			.thenReturn(synonyms());
+
+		vocabularies = VocabularyGroup.loadVocsFromIS(isLookUpService);
+		mapping = CleaningRuleMap.create(vocabularies);
+	}
+
+	@Test
+	public void testCleaning() throws Exception {
+
+		assertNotNull(vocabularies);
+		assertNotNull(mapping);
+
+		String json = IOUtils.toString(getClass().getResourceAsStream("/eu/dnetlib/dhp/oa/graph/clean/result.json"));
+		Publication p_in = MAPPER.readValue(json, Publication.class);
+
+		Publication p_out = OafCleaner.apply(p_in, mapping);
+
+		assertNotNull(p_out);
+
+		assertEquals("eng", p_out.getLanguage().getClassid());
+		assertEquals("English", p_out.getLanguage().getClassname());
+
+		assertEquals("0018", p_out.getInstance().get(0).getInstancetype().getClassid());
+		assertEquals("Annotation", p_out.getInstance().get(0).getInstancetype().getClassname());
+
+		assertEquals("CLOSED", p_out.getInstance().get(0).getAccessright().getClassid());
+		assertEquals("Closed Access", p_out.getInstance().get(0).getAccessright().getClassname());
+
+		Set<String> pidTerms = vocabularies.getTerms("dnet:pid_types");
+		assertTrue(
+			p_out
+				.getPid()
+				.stream()
+				.map(p -> p.getQualifier())
+				.allMatch(q -> pidTerms.contains(q.getClassid())));
+
+		// TODO add more assertions to verity the cleaned values
+		System.out.println(MAPPER.writeValueAsString(p_out));
+
+		/*
+		 * assertTrue( p_out .getPid() .stream() .allMatch(sp -> StringUtils.isNotBlank(sp.getValue())));
+		 */
+	}
+
+	private Stream<Qualifier> getAuthorPidTypes(Publication pub) {
+		return pub
+			.getAuthor()
+			.stream()
+			.map(a -> a.getPid())
+			.flatMap(p -> p.stream())
+			.map(s -> s.getQualifier());
+	}
+
+	private List<String> vocs() throws IOException {
+		return IOUtils
+			.readLines(CleaningFunctionTest.class.getResourceAsStream("/eu/dnetlib/dhp/oa/graph/clean/terms.txt"));
+	}
+
+	private List<String> synonyms() throws IOException {
+		return IOUtils
+			.readLines(CleaningFunctionTest.class.getResourceAsStream("/eu/dnetlib/dhp/oa/graph/clean/synonyms.txt"));
+	}
+}

dhp-workflows/dhp-graph-mapper/src/test/resources/eu/dnetlib/dhp/oa/graph/clean/result.json

@@ -0,0 +1,757 @@
+{
+  "author": [
+    {
+      "affiliation": [
+      ],
+      "fullname": "Brien, Tom",
+      "name": "Tom",
+      "pid": [
+        {
+          "dataInfo": {
+            "deletedbyinference": false,
+            "inferenceprovenance": "",
+            "inferred": false,
+            "invisible": false,
+            "provenanceaction": {
+              "classid": "sysimport:crosswalk:datasetarchive",
+              "classname": "sysimport:crosswalk:datasetarchive",
+              "schemeid": "dnet:provenanceActions",
+              "schemename": "dnet:provenanceActions"
+            },
+            "trust": "0.9"
+          },
+          "qualifier": {
+            "classid": "ORCID12",
+            "classname": "ORCID12",
+            "schemeid": "dnet:pid_types",
+            "schemename": "dnet:pid_types"
+          },
+          "value": "0000-0001-9613-6639"
+        }
+      ],
+      "rank": 1,
+      "surname": "Brien"
+    },
+    {
+      "affiliation": [
+      ],
+      "fullname": "Ade, Peter",
+      "name": "Peter",
+      "pid": [
+        {
+          "dataInfo": {
+            "deletedbyinference": false,
+            "inferenceprovenance": "",
+            "inferred": false,
+            "invisible": false,
+            "provenanceaction": {
+              "classid": "sysimport:crosswalk:datasetarchive",
+              "classname": "sysimport:crosswalk:datasetarchive",
+              "schemeid": "dnet:provenanceActions",
+              "schemename": "dnet:provenanceActions"
+            },
+            "trust": "0.9"
+          },
+          "qualifier": {
+            "classid": "xyz",
+            "classname": "XYZ",
+            "schemeid": "dnet:pid_types",
+            "schemename": "dnet:pid_types"
+          },
+          "value": "qwerty"
+        }
+      ],
+      "rank": 2,
+      "surname": "Ade"
+    },
+    {
+      "affiliation": [
+      ],
+      "fullname": "Barry, Peter S.",
+      "name": "Peter S.",
+      "pid": [
+      ],
+      "rank": 3,
+      "surname": "Barry"
+    },
+    {
+      "affiliation": [
+      ],
+      "fullname": "Dunscombe, Chris J.",
+      "name": "Chris J.",
+      "pid": [
+      ],
+      "rank": 4,
+      "surname": "Dunscombe"
+    },
+    {
+      "affiliation": [
+      ],
+      "fullname": "Leadley, David R.",
+      "name": "David R.",
+      "pid": [
+      ],
+      "rank": 5,
+      "surname": "Leadley"
+    },
+    {
+      "affiliation": [
+      ],
+      "fullname": "Morozov, Dmitry V.",
+      "name": "Dmitry V.",
+      "pid": [
+      ],
+      "rank": 6,
+      "surname": "Morozov"
+    },
+    {
+      "affiliation": [
+      ],
+      "fullname": "Myronov, Maksym",
+      "name": "Maksym",
+      "pid": [
+      ],
+      "rank": 7,
+      "surname": "Myronov"
+    },
+    {
+      "affiliation": [
+      ],
+      "fullname": "Parker, Evan",
+      "name": "Evan",
+      "pid": [
+      ],
+      "rank": 8,
+      "surname": "Parker"
+    },
+    {
+      "affiliation": [
+      ],
+      "fullname": "Prest, Martin J.",
+      "name": "Martin J.",
+      "pid": [
+      ],
+      "rank": 9,
+      "surname": "Prest"
+    },
+    {
+      "affiliation": [
+      ],
+      "fullname": "Prunnila, Mika",
+      "name": "Mika",
+      "pid": [
+      ],
+      "rank": 10,
+      "surname": "Prunnila"
+    },
+    {
+      "affiliation": [
+      ],
+      "fullname": "Sudiwala, Rashmi V.",
+      "name": "Rashmi V.",
+      "pid": [
+      ],
+      "rank": 11,
+      "surname": "Sudiwala"
+    },
+    {
+      "affiliation": [
+      ],
+      "fullname": "Whall, Terry E.",
+      "name": "Terry E.",
+      "pid": [
+      ],
+      "rank": 12,
+      "surname": "Whall"
+    },
+    {
+      "affiliation": [
+      ],
+      "fullname": "Mauskopf",
+      "name": "",
+      "pid": [
+      ],
+      "rank": 13,
+      "surname": ""
+    },
+    {
+      "affiliation": [
+      ],
+      "fullname": " P. D. ",
+      "name": "",
+      "pid": [
+      ],
+      "rank": 14,
+      "surname": ""
+    }
+  ],
+  "bestaccessright": {
+    "classid": "CLOSED",
+    "classname": "Closed Access",
+    "schemeid": "dnet:access_modes",
+    "schemename": "dnet:access_modes"
+  },
+  "collectedfrom": [
+    {
+      "key": "10|CSC_________::a2b9ce8435390bcbfc05f3cae3948747",
+      "value": "VIRTA"
+    }
+  ],
+  "context": [
+  ],
+  "contributor": [
+  ],
+  "country": [
+  ],
+  "coverage": [
+  ],
+  "dataInfo": {
+    "deletedbyinference": false,
+    "inferenceprovenance": "",
+    "inferred": false,
+    "invisible": false,
+    "provenanceaction": {
+      "classid": "sysimport:crosswalk:datasetarchive",
+      "classname": "sysimport:crosswalk:datasetarchive",
+      "schemeid": "dnet:provenanceActions",
+      "schemename": "dnet:provenanceActions"
+    },
+    "trust": "0.9"
+  },
+  "dateofacceptance": {
+    "dataInfo": {
+      "deletedbyinference": false,
+      "inferenceprovenance": "",
+      "inferred": false,
+      "invisible": false,
+      "provenanceaction": {
+        "classid": "sysimport:crosswalk:datasetarchive",
+        "classname": "sysimport:crosswalk:datasetarchive",
+        "schemeid": "dnet:provenanceActions",
+        "schemename": "dnet:provenanceActions"
+      },
+      "trust": "0.9"
+    },
+    "value": "2016-01-01"
+  },
+  "dateofcollection": "",
+  "dateoftransformation": "2020-04-22T12:34:08.009Z",
+  "description": [
+  ],
+  "externalReference": [
+  ],
+  "extraInfo": [
+  ],
+  "format": [
+  ],
+  "fulltext": [
+  ],
+  "id": "50|CSC_________::2250a70c903c6ac6e4c01438259e9375",
+  "instance": [
+    {
+      "accessright": {
+        "classid": "CLOSED",
+        "classname": "CLOSED",
+        "schemeid": "dnet:access_modes",
+        "schemename": "dnet:access_modes"
+      },
+      "collectedfrom": {
+        "key": "10|CSC_________::a2b9ce8435390bcbfc05f3cae3948747",
+        "value": "VIRTA"
+      },
+      "dateofacceptance": {
+        "dataInfo": {
+          "deletedbyinference": false,
+          "inferenceprovenance": "",
+          "inferred": false,
+          "invisible": false,
+          "provenanceaction": {
+            "classid": "sysimport:crosswalk:datasetarchive",
+            "classname": "sysimport:crosswalk:datasetarchive",
+            "schemeid": "dnet:provenanceActions",
+            "schemename": "dnet:provenanceActions"
+          },
+          "trust": "0.9"
+        },
+        "value": "2016-01-01"
+      },
+      "distributionlocation": "",
+      "hostedby": {
+        "key": "10|CSC_________::a2b9ce8435390bcbfc05f3cae3948747",
+        "value": "VIRTA"
+      },
+      "instancetype": {
+        "classid": "Comentario",
+        "classname": "Comentario",
+        "schemeid": "dnet:publication_resource",
+        "schemename": "dnet:publication_resource"
+      },
+      "url": [
+        "http://juuli.fi/Record/0275158616",
+        "http://dx.doi.org/10.1007/s109090161569x"
+      ]
+    }
+  ],
+  "journal": {
+    "dataInfo": {
+      "deletedbyinference": false,
+      "inferenceprovenance": "",
+      "inferred": false,
+      "invisible": false,
+      "provenanceaction": {
+        "classid": "sysimport:crosswalk:datasetarchive",
+        "classname": "sysimport:crosswalk:datasetarchive",
+        "schemeid": "dnet:provenanceActions",
+        "schemename": "dnet:provenanceActions"
+      },
+      "trust": "0.9"
+    },
+    "edition": "",
+    "ep": " 7",
+    "iss": "9 March",
+    "issnLinking": "",
+    "issnOnline": "",
+    "issnPrinted": "0022-2291",
+    "name": "Journal of Low Temperature Physics - Early Acces",
+    "sp": "1 ",
+    "vol": ""
+  },
+  "language": {
+    "classid": "en",
+    "classname": "en",
+    "schemeid": "dnet:languages",
+    "schemename": "dnet:languages"
+  },
+  "lastupdatetimestamp": 1591283286319,
+  "oaiprovenance": {
+    "originDescription": {
+      "altered": true,
+      "baseURL": "https%3A%2F%2Fvirta-jtp.csc.fi%2Fapi%2Fcerif",
+      "datestamp": "2019-07-30",
+      "harvestDate": "2020-04-22T11:04:38.685Z",
+      "identifier": "oai:virta-jtp.csc.fi:Publications/0275158616",
+      "metadataNamespace": ""
+    }
+  },
+  "originalId": [
+    "CSC_________::2250a70c903c6ac6e4c01438259e9375"
+  ],
+  "pid": [
+    {
+      "dataInfo": {
+        "deletedbyinference": false,
+        "inferenceprovenance": "",
+        "inferred": false,
+        "invisible": false,
+        "provenanceaction": {
+          "classid": "sysimport:crosswalk:datasetarchive",
+          "classname": "sysimport:crosswalk:datasetarchive",
+          "schemeid": "dnet:provenanceActions",
+          "schemename": "dnet:provenanceActions"
+        },
+        "trust": "0.9"
+      },
+      "qualifier": {
+        "classid": "doi",
+        "classname": "doi",
+        "schemeid": "dnet:pid_types",
+        "schemename": "dnet:pid_types"
+      },
+      "value": "10.1007/s109090161569x"
+    },
+    {
+      "dataInfo": {
+        "deletedbyinference": false,
+        "inferenceprovenance": "",
+        "inferred": false,
+        "invisible": false,
+        "provenanceaction": {
+          "classid": "sysimport:crosswalk:datasetarchive",
+          "classname": "sysimport:crosswalk:datasetarchive",
+          "schemeid": "dnet:provenanceActions",
+          "schemename": "dnet:provenanceActions"
+        },
+        "trust": "0.9"
+      },
+      "qualifier": {
+        "classid": "doi",
+        "classname": "doi",
+        "schemeid": "dnet:pid_types",
+        "schemename": "dnet:pid_types"
+      },
+      "value": "10.1007/s109090161569x"
+    },
+    {
+      "dataInfo": {
+        "deletedbyinference": false,
+        "inferenceprovenance": "",
+        "inferred": false,
+        "invisible": false,
+        "provenanceaction": {
+          "classid": "sysimport:crosswalk:datasetarchive",
+          "classname": "sysimport:crosswalk:datasetarchive",
+          "schemeid": "dnet:provenanceActions",
+          "schemename": "dnet:provenanceActions"
+        },
+        "trust": "0.9"
+      },
+      "qualifier": {
+        "classid": "doi",
+        "classname": "doi",
+        "schemeid": "dnet:pid_types",
+        "schemename": "dnet:pid_types"
+      },
+      "value": ""
+    }
+  ],
+  "relevantdate": [
+  ],
+  "resourcetype": {
+    "classid": "0001",
+    "classname": "0001",
+    "schemeid": "dnet:dataCite_resource",
+    "schemename": "dnet:dataCite_resource"
+  },
+  "resulttype": {
+    "classid": "publication",
+    "classname": "publication",
+    "schemeid": "dnet:result_typologies",
+    "schemename": "dnet:result_typologies"
+  },
+  "source": [
+  ],
+  "subject": [
+    {
+      "dataInfo": {
+        "deletedbyinference": false,
+        "inferenceprovenance": "",
+        "inferred": false,
+        "invisible": false,
+        "provenanceaction": {
+          "classid": "sysimport:crosswalk:datasetarchive",
+          "classname": "sysimport:crosswalk:datasetarchive",
+          "schemeid": "dnet:provenanceActions",
+          "schemename": "dnet:provenanceActions"
+        },
+        "trust": "0.9"
+      },
+      "qualifier": {
+        "classid": "",
+        "classname": "",
+        "schemeid": "",
+        "schemename": ""
+      },
+      "value": "ta213"
+    },
+    {
+      "dataInfo": {
+        "deletedbyinference": false,
+        "inferenceprovenance": "",
+        "inferred": false,
+        "invisible": false,
+        "provenanceaction": {
+          "classid": "sysimport:crosswalk:datasetarchive",
+          "classname": "sysimport:crosswalk:datasetarchive",
+          "schemeid": "dnet:provenanceActions",
+          "schemename": "dnet:provenanceActions"
+        },
+        "trust": "0.9"
+      },
+      "qualifier": {
+        "classid": "",
+        "classname": "",
+        "schemeid": "",
+        "schemename": ""
+      },
+      "value": "infrared detectors"
+    },
+    {
+      "dataInfo": {
+        "deletedbyinference": false,
+        "inferenceprovenance": "",
+        "inferred": false,
+        "invisible": false,
+        "provenanceaction": {
+          "classid": "sysimport:crosswalk:datasetarchive",
+          "classname": "sysimport:crosswalk:datasetarchive",
+          "schemeid": "dnet:provenanceActions",
+          "schemename": "dnet:provenanceActions"
+        },
+        "trust": "0.9"
+      },
+      "qualifier": {
+        "classid": "",
+        "classname": "",
+        "schemeid": "",
+        "schemename": ""
+      },
+      "value": "lens antennas"
+    },
+    {
+      "dataInfo": {
+        "deletedbyinference": false,
+        "inferenceprovenance": "",
+        "inferred": false,
+        "invisible": false,
+        "provenanceaction": {
+          "classid": "sysimport:crosswalk:datasetarchive",
+          "classname": "sysimport:crosswalk:datasetarchive",
+          "schemeid": "dnet:provenanceActions",
+          "schemename": "dnet:provenanceActions"
+        },
+        "trust": "0.9"
+      },
+      "qualifier": {
+        "classid": "",
+        "classname": "",
+        "schemeid": "",
+        "schemename": ""
+      },
+      "value": "silicon"
+    },
+    {
+      "dataInfo": {
+        "deletedbyinference": false,
+        "inferenceprovenance": "",
+        "inferred": false,
+        "invisible": false,
+        "provenanceaction": {
+          "classid": "sysimport:crosswalk:datasetarchive",
+          "classname": "sysimport:crosswalk:datasetarchive",
+          "schemeid": "dnet:provenanceActions",
+          "schemename": "dnet:provenanceActions"
+        },
+        "trust": "0.9"
+      },
+      "qualifier": {
+        "classid": "",
+        "classname": "",
+        "schemeid": "",
+        "schemename": ""
+      },
+      "value": "slot antennas"
+    },
+    {
+      "dataInfo": {
+        "deletedbyinference": false,
+        "inferenceprovenance": "",
+        "inferred": false,
+        "invisible": false,
+        "provenanceaction": {
+          "classid": "sysimport:crosswalk:datasetarchive",
+          "classname": "sysimport:crosswalk:datasetarchive",
+          "schemeid": "dnet:provenanceActions",
+          "schemename": "dnet:provenanceActions"
+        },
+        "trust": "0.9"
+      },
+      "qualifier": {
+        "classid": "",
+        "classname": "",
+        "schemeid": "",
+        "schemename": ""
+      },
+      "value": "strained silicon"
+    },
+    {
+      "dataInfo": {
+        "deletedbyinference": false,
+        "inferenceprovenance": "",
+        "inferred": false,
+        "invisible": false,
+        "provenanceaction": {
+          "classid": "sysimport:crosswalk:datasetarchive",
+          "classname": "sysimport:crosswalk:datasetarchive",
+          "schemeid": "dnet:provenanceActions",
+          "schemename": "dnet:provenanceActions"
+        },
+        "trust": "0.9"
+      },
+      "qualifier": {
+        "classid": "",
+        "classname": "",
+        "schemeid": "",
+        "schemename": ""
+      },
+      "value": "cold electron bolometers"
+    },
+    {
+      "dataInfo": {
+        "deletedbyinference": false,
+        "inferenceprovenance": "",
+        "inferred": false,
+        "invisible": false,
+        "provenanceaction": {
+          "classid": "sysimport:crosswalk:datasetarchive",
+          "classname": "sysimport:crosswalk:datasetarchive",
+          "schemeid": "dnet:provenanceActions",
+          "schemename": "dnet:provenanceActions"
+        },
+        "trust": "0.9"
+      },
+      "qualifier": {
+        "classid": "",
+        "classname": "",
+        "schemeid": "",
+        "schemename": ""
+      },
+      "value": "doped silicon"
+    },
+    {
+      "dataInfo": {
+        "deletedbyinference": false,
+        "inferenceprovenance": "",
+        "inferred": false,
+        "invisible": false,
+        "provenanceaction": {
+          "classid": "sysimport:crosswalk:datasetarchive",
+          "classname": "sysimport:crosswalk:datasetarchive",
+          "schemeid": "dnet:provenanceActions",
+          "schemename": "dnet:provenanceActions"
+        },
+        "trust": "0.9"
+      },
+      "qualifier": {
+        "classid": "",
+        "classname": "",
+        "schemeid": "",
+        "schemename": ""
+      },
+      "value": "measure noise"
+    },
+    {
+      "dataInfo": {
+        "deletedbyinference": false,
+        "inferenceprovenance": "",
+        "inferred": false,
+        "invisible": false,
+        "provenanceaction": {
+          "classid": "sysimport:crosswalk:datasetarchive",
+          "classname": "sysimport:crosswalk:datasetarchive",
+          "schemeid": "dnet:provenanceActions",
+          "schemename": "dnet:provenanceActions"
+        },
+        "trust": "0.9"
+      },
+      "qualifier": {
+        "classid": "",
+        "classname": "",
+        "schemeid": "",
+        "schemename": ""
+      },
+      "value": "noise equivalent power"
+    },
+    {
+      "dataInfo": {
+        "deletedbyinference": false,
+        "inferenceprovenance": "",
+        "inferred": false,
+        "invisible": false,
+        "provenanceaction": {
+          "classid": "sysimport:crosswalk:datasetarchive",
+          "classname": "sysimport:crosswalk:datasetarchive",
+          "schemeid": "dnet:provenanceActions",
+          "schemename": "dnet:provenanceActions"
+        },
+        "trust": "0.9"
+      },
+      "qualifier": {
+        "classid": "",
+        "classname": "",
+        "schemeid": "",
+        "schemename": ""
+      },
+      "value": "optical characterisation"
+    },
+    {
+      "dataInfo": {
+        "deletedbyinference": false,
+        "inferenceprovenance": "",
+        "inferred": false,
+        "invisible": false,
+        "provenanceaction": {
+          "classid": "sysimport:crosswalk:datasetarchive",
+          "classname": "sysimport:crosswalk:datasetarchive",
+          "schemeid": "dnet:provenanceActions",
+          "schemename": "dnet:provenanceActions"
+        },
+        "trust": "0.9"
+      },
+      "qualifier": {
+        "classid": "",
+        "classname": "",
+        "schemeid": "",
+        "schemename": ""
+      },
+      "value": "optical response"
+    },
+    {
+      "dataInfo": {
+        "deletedbyinference": false,
+        "inferenceprovenance": "",
+        "inferred": false,
+        "invisible": false,
+        "provenanceaction": {
+          "classid": "sysimport:crosswalk:datasetarchive",
+          "classname": "sysimport:crosswalk:datasetarchive",
+          "schemeid": "dnet:provenanceActions",
+          "schemename": "dnet:provenanceActions"
+        },
+        "trust": "0.9"
+      },
+      "qualifier": {
+        "classid": "",
+        "classname": "",
+        "schemeid": "",
+        "schemename": ""
+      },
+      "value": "photon noise"
+    },
+    {
+      "dataInfo": {
+        "deletedbyinference": false,
+        "inferenceprovenance": "",
+        "inferred": false,
+        "invisible": false,
+        "provenanceaction": {
+          "classid": "sysimport:crosswalk:datasetarchive",
+          "classname": "sysimport:crosswalk:datasetarchive",
+          "schemeid": "dnet:provenanceActions",
+          "schemename": "dnet:provenanceActions"
+        },
+        "trust": "0.9"
+      },
+      "qualifier": {
+        "classid": "",
+        "classname": "",
+        "schemeid": "",
+        "schemename": ""
+      },
+      "value": "silicon absorbers"
+    }
+  ],
+  "title": [
+    {
+      "dataInfo": {
+        "deletedbyinference": false,
+        "inferenceprovenance": "",
+        "inferred": false,
+        "invisible": false,
+        "provenanceaction": {
+          "classid": "sysimport:crosswalk:datasetarchive",
+          "classname": "sysimport:crosswalk:datasetarchive",
+          "schemeid": "dnet:provenanceActions",
+          "schemename": "dnet:provenanceActions"
+        },
+        "trust": "0.9"
+      },
+      "qualifier": {
+        "classid": "main title",
+        "classname": "main title",
+        "schemeid": "dnet:dataCite_title",
+        "schemename": "dnet:dataCite_title"
+      },
+      "value": "Optical response of strained- and unstrained-silicon cold-electron bolometers"
+    }
+  ]
+}

dhp-workflows/dhp-graph-provision-scholexplorer/pom.xml

@@ -3,7 +3,7 @@
     <parent>
         <artifactId>dhp-workflows</artifactId>
         <groupId>eu.dnetlib.dhp</groupId>
-        <version>1.2.2-SNAPSHOT</version>
+        <version>1.2.3-SNAPSHOT</version>
     </parent>
     <modelVersion>4.0.0</modelVersion>
 

dhp-workflows/dhp-graph-provision/pom.xml

@@ -3,7 +3,7 @@
     <parent>
         <artifactId>dhp-workflows</artifactId>
         <groupId>eu.dnetlib.dhp</groupId>
-        <version>1.2.2-SNAPSHOT</version>
+        <version>1.2.3-SNAPSHOT</version>
     </parent>
     <modelVersion>4.0.0</modelVersion>
 

dhp-workflows/dhp-graph-provision/src/main/java/eu/dnetlib/dhp/oa/provision/utils/XmlRecordFactory.java

@@ -897,7 +897,10 @@ public class XmlRecordFactory implements Serializable {
 					metadata.add(XmlSerializationUtils.mapQualifier("contracttype", p.getContracttype()));
 				}
 				if (p.getOamandatepublications() != null) {
-					metadata.add(XmlSerializationUtils.asXmlElement("oamandatepublications", p.getOamandatepublications().getValue()));
+					metadata
+						.add(
+							XmlSerializationUtils
+								.asXmlElement("oamandatepublications", p.getOamandatepublications().getValue()));
 				}
 				if (p.getEcsc39() != null) {
 					metadata.add(XmlSerializationUtils.asXmlElement("ecsc39", p.getEcsc39().getValue()));
@@ -1168,10 +1171,10 @@ public class XmlRecordFactory implements Serializable {
 									.asXmlElement(
 										"distributionlocation", instance.getDistributionlocation()));
 					}
-					if (instance.getRefereed() != null && isNotBlank(instance.getRefereed().getValue())) {
+					if (instance.getRefereed() != null && !instance.getRefereed().isBlank()) {
 						fields
 							.add(
-								XmlSerializationUtils.asXmlElement("refereed", instance.getRefereed().getValue()));
+								XmlSerializationUtils.mapQualifier("refereed", instance.getRefereed()));
 					}
 					if (instance.getProcessingchargeamount() != null
 						&& isNotBlank(instance.getProcessingchargeamount().getValue())) {

dhp-workflows/dhp-stats-update/pom.xml

@@ -3,7 +3,7 @@
     <parent>
         <artifactId>dhp-workflows</artifactId>
         <groupId>eu.dnetlib.dhp</groupId>
-        <version>1.2.2-SNAPSHOT</version>
+        <version>1.2.3-SNAPSHOT</version>
     </parent>
     <modelVersion>4.0.0</modelVersion>
     <artifactId>dhp-stats-update</artifactId>

dhp-workflows/dhp-worfklow-profiles/pom.xml

@@ -3,7 +3,7 @@
     <parent>
         <artifactId>dhp-workflows</artifactId>
         <groupId>eu.dnetlib.dhp</groupId>
-        <version>1.2.2-SNAPSHOT</version>
+        <version>1.2.3-SNAPSHOT</version>
     </parent>
     <modelVersion>4.0.0</modelVersion>
 

dhp-workflows/pom.xml

@@ -6,7 +6,7 @@
     <parent>
         <groupId>eu.dnetlib.dhp</groupId>
         <artifactId>dhp</artifactId>
-        <version>1.2.2-SNAPSHOT</version>
+        <version>1.2.3-SNAPSHOT</version>
         <relativePath>../</relativePath>
     </parent>
 

pom.xml

@@ -3,7 +3,7 @@
 	<modelVersion>4.0.0</modelVersion>
 	<groupId>eu.dnetlib.dhp</groupId>
 	<artifactId>dhp</artifactId>
-	<version>1.2.2-SNAPSHOT</version>
+	<version>1.2.3-SNAPSHOT</version>
 	<packaging>pom</packaging>
 
 	<licenses>