deleting events

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

@@ -31,6 +31,10 @@
 			<artifactId>elasticsearch-hadoop</artifactId>
 		</dependency>
 
+		<dependency>
+			<groupId>org.apache.httpcomponents</groupId>
+			<artifactId>httpclient</artifactId>
+		</dependency>
 
 		<dependency>
 			<groupId>eu.dnetlib.dhp</groupId>

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

@@ -7,6 +7,10 @@ import java.util.Map;
 
 import org.apache.commons.io.IOUtils;
 import org.apache.commons.lang3.math.NumberUtils;
+import org.apache.http.client.methods.CloseableHttpResponse;
+import org.apache.http.client.methods.HttpDelete;
+import org.apache.http.impl.client.CloseableHttpClient;
+import org.apache.http.impl.client.HttpClients;
 import org.apache.spark.SparkConf;
 import org.apache.spark.api.java.JavaRDD;
 import org.apache.spark.sql.Dataset;
@@ -54,6 +58,9 @@ public class IndexEventSubsetJob {
 		final int maxEventsForTopic = NumberUtils.toInt(parser.get("maxEventsForTopic"));
 		log.info("maxEventsForTopic: {}", maxEventsForTopic);
 
+		final String brokerApiBaseUrl = parser.get("brokerApiBaseUrl");
+		log.info("brokerApiBaseUrl: {}", brokerApiBaseUrl);
+
 		final SparkSession spark = SparkSession.builder().config(conf).getOrCreate();
 
 		final TypedColumn<Event, EventGroup> aggr = new EventSubsetAggregator(maxEventsForTopic).toColumn();
@@ -84,7 +91,26 @@ public class IndexEventSubsetJob {
 		esCfg.put("es.batch.size.entries", "200");
 		esCfg.put("es.nodes.wan.only", "true");
 
+		log.info("*** Start indexing");
 		JavaEsSpark.saveJsonToEs(inputRdd, index, esCfg);
+		log.info("*** End indexing");
+
+		log.info("*** Deleting old events");
+		final String message = deleteOldEvents(brokerApiBaseUrl, now - 1000);
+		log.info("*** Deleted events: " + message);
+
+	}
+
+	private static String deleteOldEvents(final String brokerApiBaseUrl, final long l) throws Exception {
+		final String url = brokerApiBaseUrl + "/api/events/byCreationDate/0/" + l;
+		final HttpDelete req = new HttpDelete(url);
+
+		try (final CloseableHttpClient client = HttpClients.createDefault()) {
+			try (final CloseableHttpResponse response = client.execute(req)) {
+				return IOUtils.toString(response.getEntity().getContent());
+			}
+		}
+
 	}
 
 	private static String prepareEventForIndexing(final Event e, final long creationDate, final LongAccumulator acc)

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

@@ -36,6 +36,10 @@
         	<name>maxIndexedEventsForDsAndTopic</name>
         	<description>the max number of events for each couple (ds/topic)</description>
         </property>
+        <property>
+        	<name>brokerApiBaseUrl</name>
+        	<description>the url of the broker service api</description>
+        </property>
         <property>
             <name>sparkDriverMemory</name>
             <description>memory for driver process</description>
@@ -452,6 +456,7 @@
             <arg>--index</arg><arg>${esEventIndexName}</arg>
             <arg>--esHost</arg><arg>${esIndexHost}</arg>
             <arg>--maxEventsForTopic</arg><arg>${maxIndexedEventsForDsAndTopic}</arg>
+            <arg>--brokerApiBaseUrl</arg><arg>${brokerApiBaseUrl}</arg>
         </spark>
         <ok to="stats"/>
         <error to="Kill"/>

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

@@ -22,5 +22,11 @@
 		"paramLongName": "maxEventsForTopic",
 		"paramDescription": "the max number of events for each couple (ds/topic)",
 		"paramRequired": true
+	},
+	{
+		"paramName": "broker",
+		"paramLongName": "brokerApiBaseUrl",
+		"paramDescription": "the url of the broker service api",
+		"paramRequired": true
 	}
 ]

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

@@ -36,6 +36,10 @@
         	<name>maxIndexedEventsForDsAndTopic</name>
         	<description>the max number of events for each couple (ds/topic)</description>
         </property>
+        <property>
+        	<name>brokerApiBaseUrl</name>
+        	<description>the url of the broker service api</description>
+        </property>
         <property>
             <name>sparkDriverMemory</name>
             <description>memory for driver process</description>
@@ -118,6 +122,7 @@
             <arg>--index</arg><arg>${esEventIndexName}</arg>
             <arg>--esHost</arg><arg>${esIndexHost}</arg>
             <arg>--maxEventsForTopic</arg><arg>${maxIndexedEventsForDsAndTopic}</arg>
+            <arg>--brokerApiBaseUrl</arg><arg>${brokerApiBaseUrl}</arg>
         </spark>
         <ok to="End"/>
         <error to="Kill"/>