UNOMI-967: Harden clustered scheduler locks, recovery, and tests

api/src/main/java/org/apache/unomi/api/EventInfo.java

@@ -17,12 +17,16 @@
 
 package org.apache.unomi.api;
 
+import java.io.Serializable;
+
 /**
  * One event type entry in {@link ServerInfo#getEventTypes()}.
  * Pairs the event type name with how many matching events exist on the server,
  * giving operators and clients a quick view of which event types are active.
  */
-public class EventInfo {
+public class EventInfo implements Serializable {
+
+    private static final long serialVersionUID = 1L;
 
     private String name;
     private Long occurences;

api/src/main/java/org/apache/unomi/api/ServerInfo.java

@@ -17,6 +17,7 @@
 
 package org.apache.unomi.api;
 
+import java.io.Serializable;
 import java.util.ArrayList;
 import java.util.Date;
 import java.util.List;
@@ -28,7 +29,9 @@
  * operators, and other nodes can read version, build metadata, supported
  * {@link EventInfo} types, optional capability flags, and banner logo lines.
  */
-public class ServerInfo {
+public class ServerInfo implements Serializable {
+
+    private static final long serialVersionUID = 1L;
 
     /** @api.example unomi */
     private String serverIdentifier;

api/src/main/java/org/apache/unomi/api/services/SchedulerService.java

@@ -351,11 +351,22 @@ interface TaskBuilder {
         /**
          * Disallows parallel execution.
          * Task will use locking to ensure only one instance runs at a time.
+         * This is the default for {@link #newTask(String)}.
          *
          * @return this builder for method chaining
          */
         TaskBuilder disallowParallelExecution();
 
+        /**
+         * Allows parallel execution (disables exclusive cluster locking for this task).
+         * Use only when duplicate side effects across nodes are intentional; prefer the
+         * default exclusive behaviour for cluster-wide jobs. {@link #runOnAllNodes()}
+         * enables this automatically.
+         *
+         * @return this builder for method chaining
+         */
+        TaskBuilder allowParallelExecution();
+
         /**
          * Sets the task executor.
          *

clear-elasticsearch.sh

@@ -48,13 +48,17 @@ if [ "${_IS_SOURCED}" = false ]; then
     exit 1
 fi
 
-# Clear Elasticsearch environment variables
+# Clear Elasticsearch environment variables (must cover everything setup-elasticsearch.sh exports)
 unset UNOMI_ELASTICSEARCH_CLUSTERNAME
 unset UNOMI_ELASTICSEARCH_ADDRESSES
 unset UNOMI_ELASTICSEARCH_USERNAME
 unset UNOMI_ELASTICSEARCH_PASSWORD
 unset UNOMI_ELASTICSEARCH_SSL_ENABLE
 unset UNOMI_ELASTICSEARCH_SSL_TRUST_ALL_CERTIFICATES
+# Also set by setup-elasticsearch.sh / setup-opensearch.sh
+unset UNOMI_DISTRIBUTION
+
+unset _IS_SOURCED
 
 echo "Elasticsearch environment variables cleared."
 

clear-opensearch.sh

@@ -48,13 +48,18 @@ if [ "${_IS_SOURCED}" = false ]; then
     exit 1
 fi
 
-# Clear OpenSearch environment variables
+# Clear OpenSearch environment variables (must cover everything setup-opensearch.sh exports,
+# plus common overrides users may have set manually)
 unset UNOMI_OPENSEARCH_CLUSTERNAME
 unset UNOMI_OPENSEARCH_ADDRESSES
 unset UNOMI_OPENSEARCH_USERNAME
 unset UNOMI_OPENSEARCH_PASSWORD
 unset UNOMI_OPENSEARCH_SSL_ENABLE
 unset UNOMI_OPENSEARCH_SSL_TRUST_ALL_CERTIFICATES
+# Also set by setup-opensearch.sh / setup-elasticsearch.sh
+unset UNOMI_DISTRIBUTION
+
+unset _IS_SOURCED
 
 echo "OpenSearch environment variables cleared."
 

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

@@ -669,6 +669,13 @@ public Option[] config() {
                 editConfigurationFilePut("etc/custom.system.properties", "org.apache.unomi.opensearch.rollover.maxDocs", "300"),
                 editConfigurationFilePut("etc/custom.system.properties", "org.apache.unomi.opensearch.minimalClusterState", "YELLOW"),
                 editConfigurationFilePut("etc/custom.system.properties", "org.apache.unomi.migration.tenant.id", TEST_TENANT_ID),
+                // Default scheduler.thread.poolSize (5) is sized for the near-instant in-memory unit-test
+                // double, not a real ES/OS backend. Under real refresh/write latency, the checker, task
+                // executions, and lease-renewal heartbeats (see scheduler.adoc) compete for the same small
+                // pool; when it saturates, heartbeats starve, locks age out from under live executions, and
+                // crash recovery repeatedly (mis)reclaims them, logging "Lock verification failed... after
+                // CAS" every checker tick. Widen it for ITs so heartbeats/checker never starve.
+                editConfigurationFilePut("etc/custom.system.properties", "org.apache.unomi.scheduler.thread.poolSize", "10"),
 
                 systemProperty("org.ops4j.pax.exam.rbc.rmi.port").value("1199"),
                 systemProperty("org.apache.unomi.healthcheck.enabled").value("true"),
@@ -747,6 +754,15 @@ public Option[] config() {
         karafOptions.add(editConfigurationFilePut("etc/org.ops4j.pax.logging.cfg", "log4j2.logger.paxStore.name", "org.ops4j.store"));
         karafOptions.add(editConfigurationFilePut("etc/org.ops4j.pax.logging.cfg", "log4j2.logger.paxStore.level", "WARN"));
 
+        // UNOMI-967 diagnostic: DEBUG-level "LOCK-DIAG"-tagged logging was added throughout the
+        // scheduler's lock acquisition/renewal/recovery paths (TaskLockManager, TaskRecoveryManager,
+        // SchedulerServiceImpl, TaskExecutionManager) to root-cause a live-IT-only bug where
+        // "Lock verification failed... after CAS" fires on every checker tick for the whole suite
+        // (a real ES backend, not the in-memory unit-test double). Bump this package to DEBUG for
+        // ITs only so that logging is actually captured. TODO: remove once root-caused and fixed.
+        karafOptions.add(editConfigurationFilePut("etc/org.ops4j.pax.logging.cfg", "log4j2.logger.schedulerDiag.name", "org.apache.unomi.services.impl.scheduler"));
+        karafOptions.add(editConfigurationFilePut("etc/org.ops4j.pax.logging.cfg", "log4j2.logger.schedulerDiag.level", "DEBUG"));
+
         // Enable debug logging for Karaf Resolver to diagnose bundle refresh issues (default: disabled)
         boolean enableResolverDebug = Boolean.parseBoolean(System.getProperty(RESOLVER_DEBUG_PROPERTY, "false"));
         if (enableResolverDebug) {