From 961a0d0b4952a51aaf27fb10c3a6a2bcba018710 Mon Sep 17 00:00:00 2001
From: Claudio Atzori <claudio.atzori@isti.cnr.it>
Date: Mon, 22 Jun 2020 10:20:45 +0200
Subject: [PATCH] [actionset promotion] log debugging info in case of error in
 the action payload extraction or parsing the data

---
 .../PromoteActionPayloadForGraphTableJob.java | 29 +++++++++++++++----
 1 file changed, 23 insertions(+), 6 deletions(-)

diff --git a/dhp-workflows/dhp-actionmanager/src/main/java/eu/dnetlib/dhp/actionmanager/promote/PromoteActionPayloadForGraphTableJob.java b/dhp-workflows/dhp-actionmanager/src/main/java/eu/dnetlib/dhp/actionmanager/promote/PromoteActionPayloadForGraphTableJob.java
index 17bfc4af3..5fa9e6723 100644
--- a/dhp-workflows/dhp-actionmanager/src/main/java/eu/dnetlib/dhp/actionmanager/promote/PromoteActionPayloadForGraphTableJob.java
+++ b/dhp-workflows/dhp-actionmanager/src/main/java/eu/dnetlib/dhp/actionmanager/promote/PromoteActionPayloadForGraphTableJob.java
@@ -4,6 +4,7 @@ package eu.dnetlib.dhp.actionmanager.promote;
 import static eu.dnetlib.dhp.common.SparkSessionSupport.runWithSparkSession;
 import static eu.dnetlib.dhp.schema.common.ModelSupport.isSubClass;
 
+import java.io.IOException;
 import java.util.Objects;
 import java.util.Optional;
 import java.util.function.BiFunction;
@@ -20,6 +21,7 @@ import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
 import com.fasterxml.jackson.databind.ObjectMapper;
+import com.fasterxml.jackson.databind.exc.UnrecognizedPropertyException;
 
 import eu.dnetlib.dhp.application.ArgumentApplicationParser;
 import eu.dnetlib.dhp.common.FunctionalInterfaceSupport.SerializableSupplier;
@@ -134,24 +136,39 @@ public class PromoteActionPayloadForGraphTableJob {
 			.map(
 				(MapFunction<String, G>) value -> OBJECT_MAPPER.readValue(value, rowClazz),
 				Encoders.bean(rowClazz));
-
-		/*
-		 * return spark .read() .parquet(path) .as(Encoders.bean(rowClazz));
-		 */
 	}
 
 	private static <A extends Oaf> Dataset<A> readActionPayload(
 		SparkSession spark, String path, Class<A> actionPayloadClazz) {
 		logger.info("Reading action payload from path: {}", path);
+
 		return spark
 			.read()
 			.parquet(path)
+			.map((MapFunction<Row, String>) value -> extractPayload(value), Encoders.STRING())
 			.map(
-				(MapFunction<Row, A>) value -> OBJECT_MAPPER
-					.readValue(value.<String> getAs("payload"), actionPayloadClazz),
+				(MapFunction<String, A>) value -> decodePayload(actionPayloadClazz, value),
 				Encoders.bean(actionPayloadClazz));
 	}
 
+	private static String extractPayload(Row value) {
+		try {
+			return value.<String> getAs("payload");
+		} catch (IllegalArgumentException | ClassCastException e) {
+			logger.error("cannot extract payload from action: {}", value.toString());
+			throw e;
+		}
+	}
+
+	private static <A extends Oaf> A decodePayload(Class<A> actionPayloadClazz, String payload) throws IOException {
+		try {
+			return OBJECT_MAPPER.readValue(payload, actionPayloadClazz);
+		} catch (UnrecognizedPropertyException e) {
+			logger.error("error decoding payload: {}", payload);
+			throw e;
+		}
+	}
+
 	private static <G extends Oaf, A extends Oaf> Dataset<G> promoteActionPayloadForGraphTable(
 		Dataset<G> rowDS,
 		Dataset<A> actionPayloadDS,