UNOMI-784: use tasks to perform update_by_query and delete_by_query

itests/pom.xml

@@ -226,6 +226,7 @@
                             </environmentVariables>
                             <instanceSettings>
                                 <properties>
+                                    <xpack.ml.enabled>false</xpack.ml.enabled>
                                     <path.repo>${project.build.directory}/snapshots_repository</path.repo>
                                     <cluster.routing.allocation.disk.threshold_enabled>false</cluster.routing.allocation.disk.threshold_enabled>
                                     <http.cors.allow-origin>*</http.cors.allow-origin>

itests/src/test/java/org/apache/unomi/itests/BaseIT.java

@@ -93,12 +93,7 @@
 import java.security.NoSuchAlgorithmException;
 import java.security.SecureRandom;
 import java.security.cert.X509Certificate;
-import java.util.ArrayList;
-import java.util.Arrays;
-import java.util.Dictionary;
-import java.util.List;
-import java.util.Map;
-import java.util.Objects;
+import java.util.*;
 import java.util.concurrent.CountDownLatch;
 import java.util.function.Predicate;
 import java.util.function.Supplier;
@@ -256,6 +251,7 @@ public Option[] config() {
                 editConfigurationFilePut("etc/custom.system.properties", "org.apache.unomi.graphql.feature.activated", "true"),
                 editConfigurationFilePut("etc/custom.system.properties", "org.apache.unomi.elasticsearch.cluster.name", "contextElasticSearchITests"),
                 editConfigurationFilePut("etc/custom.system.properties", "org.apache.unomi.elasticsearch.addresses", "localhost:9400"),
+                editConfigurationFilePut("etc/custom.system.properties", "org.apache.unomi.elasticsearch.taskWaitingPollingInterval", "50"),
 
                 systemProperty("org.ops4j.pax.exam.rbc.rmi.port").value("1199"),
                 systemProperty("org.apache.unomi.hazelcast.group.name").value("cellar"),
@@ -381,13 +377,23 @@ public void updateServices() throws InterruptedException {
         persistenceService = getService(PersistenceService.class);
         definitionsService = getService(DefinitionsService.class);
         rulesService = getService(RulesService.class);
+        segmentService = getService(SegmentService.class);
     }
 
     public void updateConfiguration(String serviceName, String configPid, String propName, Object propValue)
             throws InterruptedException, IOException {
+        Map<String, Object> props = new HashMap<>();
+        props.put(propName, propValue);
+        updateConfiguration(serviceName, configPid, props);
+    }
+
+    public void updateConfiguration(String serviceName, String configPid, Map<String, Object> propsToSet)
+            throws InterruptedException, IOException {
         org.osgi.service.cm.Configuration cfg = configurationAdmin.getConfiguration(configPid);
         Dictionary<String, Object> props = cfg.getProperties();
-        props.put(propName, propValue);
+        for (Map.Entry<String, Object> propToSet : propsToSet.entrySet()) {
+            props.put(propToSet.getKey(), propToSet.getValue());
+        }
 
         waitForReRegistration(serviceName, () -> {
             try {

itests/src/test/java/org/apache/unomi/itests/ProfileServiceIT.java

@@ -155,8 +155,17 @@ public void testGetProfileWithWrongScrollerIdThrowException()
             throws InterruptedException, NoSuchFieldException, IllegalAccessException, IOException {
         boolean throwExceptionCurrent = false;
         Configuration elasticSearchConfiguration = configurationAdmin.getConfiguration("org.apache.unomi.persistence.elasticsearch");
-        if (elasticSearchConfiguration != null) {
-            throwExceptionCurrent = Boolean.getBoolean((String) elasticSearchConfiguration.getProperties().get("throwExceptions"));
+        if (elasticSearchConfiguration != null && elasticSearchConfiguration.getProperties().get("throwExceptions") != null) {
+            try {
+                if (elasticSearchConfiguration.getProperties().get("throwExceptions") instanceof String) {
+                    throwExceptionCurrent = Boolean.parseBoolean((String) elasticSearchConfiguration.getProperties().get("throwExceptions"));
+                } else {
+                    // already a boolean
+                    throwExceptionCurrent = (Boolean) elasticSearchConfiguration.getProperties().get("throwExceptions");
+                }
+            } catch (Throwable e) {
+                // Not able to cast the property
+            }
         }
 
         updateConfiguration(PersistenceService.class.getName(), "org.apache.unomi.persistence.elasticsearch", "throwExceptions", true);

itests/src/test/java/org/apache/unomi/itests/SegmentIT.java

@@ -39,18 +39,15 @@
 import org.ops4j.pax.exam.spi.reactors.ExamReactorStrategy;
 import org.ops4j.pax.exam.spi.reactors.PerSuite;
 import org.ops4j.pax.exam.util.Filter;
+import org.osgi.service.cm.Configuration;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
 import javax.inject.Inject;
 import java.text.SimpleDateFormat;
 import java.time.LocalDate;
 import java.time.ZoneId;
-import java.util.ArrayList;
-import java.util.Collections;
-import java.util.Date;
-import java.util.List;
-import java.util.Map;
+import java.util.*;
 
 @RunWith(PaxExam.class)
 @ExamReactorStrategy(PerSuite.class)

package/src/main/resources/etc/custom.system.properties

@@ -132,7 +132,20 @@ org.apache.unomi.elasticsearch.defaultIndex.indexMaxDocValueFieldsSearch=${env:U
 org.apache.unomi.elasticsearch.defaultQueryLimit=${env:UNOMI_ELASTICSEARCH_DEFAULTQUERYLIMIT:-10}
 org.apache.unomi.elasticsearch.aggregateQueryBucketSize=${env:UNOMI_ELASTICSEARCH_AGGREGATEBUCKETSIZE:-5000}
 org.apache.unomi.elasticsearch.maximumIdsQueryCount=${env:UNOMI_ELASTICSEARCH_MAXIMUMIDSQUERYCOUNT:-5000}
-org.apache.unomi.elasticsearch.clientSocketTimeout=${env:UNOMI_ELASTICSEARCH_CLIENT_SOCKET_TIMEOUT:-}
+# Defines the socket timeout (SO_TIMEOUT) in milliseconds, which is the timeout for waiting for data or, put differently, a maximum period inactivity between two consecutive data packets).
+# A timeout value of zero is interpreted as an infinite timeout. A negative value is interpreted as undefined (system default).
+# Default: -1 (System default)
+org.apache.unomi.elasticsearch.clientSocketTimeout=${env:UNOMI_ELASTICSEARCH_CLIENT_SOCKET_TIMEOUT:--1}
+# Defines the waiting for task completion timeout in milliseconds.
+# Some operations like update_by_query and delete_by_query are delegated to ElasticSearch using tasks
+# For consistency the thread that trigger one of those operations will wait for the task to be completed on ElasticSearch side.
+# This timeout configuration is here to ensure not blocking the thread infinitely, in case of very long running tasks.
+# A timeout value of zero or negative is interpreted as an infinite timeout.
+# Default: 3600000 (one hour)
+org.apache.unomi.elasticsearch.taskWaitingTimeout=${env:UNOMI_ELASTICSEARCH_TASK_WAITING_TIMEOUT:-3600000}
+# Defines the polling interval in milliseconds, which is used to check if task is completed on ElasticSearch side
+# Default: 1000 (1 second)
+org.apache.unomi.elasticsearch.taskWaitingPollingInterval=${env:UNOMI_ELASTICSEARCH_TASK_WAITING_POLLING_INTERVAL:-1000}
 org.apache.unomi.elasticsearch.pastEventsDisablePartitions=${env:UNOMI_ELASTICSEARCH_PAST_EVENTS_DISABLE_PARTITIONS:-false}
 org.apache.unomi.elasticsearch.aggQueryThrowOnMissingDocs=${env:UNOMI_ELASTICSEARCH_AGG_QUERY_THROW_ON_MISSING_DOCS:-false}
 org.apache.unomi.elasticsearch.aggQueryMaxResponseSizeHttp=${env:UNOMI_ELASTICSEARCH_AGG_QUERY_MAX_RESPONSE_SIZE_HTTP:-}