From fae681fea1812513f900ad6457455dad9c1f80bc Mon Sep 17 00:00:00 2001
From: "miriam.baglioni" <miriam.baglioni@isti.cnr.it>
Date: Sun, 3 Jul 2022 17:39:58 +0200
Subject: [PATCH] [Country Propagation] add check to avoid NPE on
 datasource.getDatasourceType().getClassis()

---
 .../PrepareDatasourceCountryAssociation.java                   | 3 +++
 1 file changed, 3 insertions(+)

diff --git a/dhp-workflows/dhp-enrichment/src/main/java/eu/dnetlib/dhp/countrypropagation/PrepareDatasourceCountryAssociation.java b/dhp-workflows/dhp-enrichment/src/main/java/eu/dnetlib/dhp/countrypropagation/PrepareDatasourceCountryAssociation.java
index 42b7804ea..b9f3bff52 100644
--- a/dhp-workflows/dhp-enrichment/src/main/java/eu/dnetlib/dhp/countrypropagation/PrepareDatasourceCountryAssociation.java
+++ b/dhp-workflows/dhp-enrichment/src/main/java/eu/dnetlib/dhp/countrypropagation/PrepareDatasourceCountryAssociation.java
@@ -6,6 +6,7 @@ import static eu.dnetlib.dhp.common.SparkSessionSupport.runWithSparkSession;
 
 import java.util.Arrays;
 import java.util.List;
+import java.util.Optional;
 
 import org.apache.commons.io.IOUtils;
 import org.apache.spark.SparkConf;
@@ -87,6 +88,8 @@ public class PrepareDatasourceCountryAssociation {
 		Dataset<Datasource> datasource = readPath(spark, inputPath + "/datasource", Datasource.class)
 			.filter(
 				(FilterFunction<Datasource>) ds -> !ds.getDataInfo().getDeletedbyinference() &&
+					Optional.ofNullable(ds.getDatasourcetype()).isPresent() &&
+					Optional.ofNullable(ds.getDatasourcetype().getClassid()).isPresent() &&
 					(allowedtypes.contains(ds.getDatasourcetype().getClassid()) ||
 						whitelist.contains(ds.getId())));