This is an automated email from the ASF dual-hosted git repository.
sergehuber pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/unomi.git
The following commit(s) were added to refs/heads/master by this push:
new 0ff5e659e UNOMI-979: Judge scheduler lock expiry by the owner's
recorded lease, deflake scheduler tests (#855)
0ff5e659e is described below
commit 0ff5e659e88f9f9ba0845502d690819f97717189
Author: Serge Huber <[email protected]>
AuthorDate: Tue Aug 18 14:18:58 2026 +0200
UNOMI-979: Judge scheduler lock expiry by the owner's recorded lease,
deflake scheduler tests (#855)
Merge UNOMI-979 unit test fix
---
.github/workflows/unomi-ci-build-tests.yml | 83 +++++-
.../org/apache/unomi/api/tasks/ScheduledTask.java | 36 ++-
build.sh | 22 +-
itests/lib/it-run-memory.sh | 184 +++++++++++-
itests/sample-it-memory.sh | 9 +-
.../META-INF/cxs/mappings/scheduledTask.json | 3 +
.../META-INF/cxs/mappings/scheduledTask.json | 3 +
.../impl/scheduler/SchedulerServiceImpl.java | 3 +
.../impl/scheduler/TaskExecutionManager.java | 62 +++++
.../services/impl/scheduler/TaskLockManager.java | 29 +-
.../impl/scheduler/TaskRecoveryManager.java | 2 +
.../services/impl/scheduler/TaskStateManager.java | 2 +
.../ScheduledTaskLeaseSerializationTest.java | 124 +++++++++
.../scheduler/SchedulerDiagnosticsExtension.java | 271 ++++++++++++++++++
.../scheduler/SchedulerServiceClusterRaceTest.java | 145 ++++++++++
.../impl/scheduler/SchedulerServiceImplTest.java | 310 ++++++++++++++++-----
.../impl/scheduler/TaskExecutionManagerTest.java | 111 +++++++-
.../impl/scheduler/TaskLockManagerTest.java | 183 ++++++++++++
18 files changed, 1475 insertions(+), 107 deletions(-)
diff --git a/.github/workflows/unomi-ci-build-tests.yml
b/.github/workflows/unomi-ci-build-tests.yml
index 076ab448e..07f2f5858 100644
--- a/.github/workflows/unomi-ci-build-tests.yml
+++ b/.github/workflows/unomi-ci-build-tests.yml
@@ -21,7 +21,9 @@ jobs:
unit-tests:
name: Execute unit tests
runs-on: ubuntu-latest
- timeout-minutes: 15
+ # Slightly above the previous 15: retried failures (rerunFailingTestsCount
below) add time
+ # on a red build, and a timeout is a much worse signal than a clean
failure.
+ timeout-minutes: 20
steps:
- uses: actions/checkout@v5
- name: Set up JDK 17
@@ -36,12 +38,79 @@ jobs:
sudo apt-get install -y graphviz
dot -V
- name: Build and Unit tests
+ env:
+ # Retry a failing test twice before calling the build red. Several
suites (notably the
+ # scheduler ones) are timing-sensitive and this runner has 2 vCPU, so
a single unlucky
+ # scheduling hiccup should not fail a whole build. A test that only
passes on retry is
+ # NOT silently forgiven: Surefire records it as a flake, and the step
below surfaces
+ # every one in the job summary so the flake rate stays visible instead
of becoming
+ # invisible green.
+ MAVEN_EXTRA_OPTS: -Dsurefire.rerunFailingTestsCount=2
run: ./build.sh --ci
# Keep only third-party dependencies in the post-job Maven cache: Unomi's
own
# snapshots are rebuilt every run and would only bloat the cache / risk
staleness
- name: Clean Unomi artifacts from Maven cache
if: always()
run: rm -rf ~/.m2/repository/org/apache/unomi
+ # A flake is a test that failed and then passed on retry. The build is
green, so without
+ # this the signal is lost entirely — which is how the scheduler suites
stayed unreliable
+ # for as long as they did.
+ - name: Detect flaky tests
+ id: flakes
+ if: always()
+ run: |
+ python3 - <<'PY' >> "$GITHUB_STEP_SUMMARY"
+ import glob, os, xml.etree.ElementTree as ET
+ flaky = []
+ for path in glob.glob('**/target/surefire-reports/TEST-*.xml',
recursive=True):
+ try:
+ root = ET.parse(path).getroot()
+ except ET.ParseError:
+ continue
+ for case in root.iter('testcase'):
+ reruns = case.findall('flakyFailure') +
case.findall('flakyError')
+ if reruns:
+ msg = (reruns[0].get('message') or
'').strip().replace('\n', ' ')
+ flaky.append((case.get('classname', '?'), case.get('name',
'?'),
+ len(reruns), msg[:160]))
+ if flaky:
+ print('### :warning: Flaky tests detected\n')
+ print('These failed and then passed on retry. The build is green,
but each one is')
+ print('a real intermittent failure worth investigating.\n')
+ print('| Test | Retries | First failure |')
+ print('| --- | --- | --- |')
+ for cls, name, n, msg in sorted(flaky):
+ print(f'| `{cls}.{name}` | {n} | {msg or "—"} |')
+ else:
+ print('### No flaky tests detected\n')
+ with open(os.environ['GITHUB_OUTPUT'], 'a') as out:
+ out.write(f'found={"true" if flaky else "false"}\n')
+ out.write(f'count={len(flaky)}\n')
+ PY
+ # Uploaded when the build failed OR when something only passed on retry:
those are exactly
+ # the runs where the reports (and the scheduler diagnostics dumped into
them) are worth
+ # keeping. Skipped on a clean green run so this does not accumulate on
every push.
+ - name: Archive unit test reports
+ uses: actions/upload-artifact@v6
+ if: always() && (job.status == 'failure' || steps.flakes.outputs.found
== 'true')
+ with:
+ name: unit-test-reports-jdk17-${{ github.run_number }}
+ path: |
+ **/target/surefire-reports/**
+ if-no-files-found: ignore
+ retention-days: 14
+ # Always publish so a later "re-run failed jobs" pass updates the check to
green, matching
+ # the integration-test job's behaviour.
+ - name: Publish Test Report
+ uses: mikepenz/action-junit-report@v3
+ if: always()
+ continue-on-error: true
+ with:
+ report_paths: '**/target/surefire-reports/TEST-*.xml'
+ check_name: 'JUnit Test Report (unit tests)'
+ update_check: true
+ fail_on_failure: false
+ require_tests: false
integration-tests:
name: Execute integration tests
@@ -74,11 +143,19 @@ jobs:
MAVEN_EXTRA_OPTS: >-
-Dopensearch.port=${{ matrix.port }}
-Delasticsearch.port=${{ matrix.port }}
+ # This job is gated on `needs: unit-tests`, so the unit suite and the
Javadoc/checkstyle
+ # validation have already passed on this exact commit. Re-running
either here is pure
+ # duplication before the integration tests this job exists for, and
the legs run
+ # sequentially (max-parallel: 1), so it costs twice over.
+ # --skip-unit-tests activates the skip-unit-tests profile, which
sets surefire's skip
+ # only: failsafe, and therefore the ITs, still run.
+ # --no-javadoc drops the two extra full-reactor invocations --ci adds
+ # (javadoc:javadoc and javadoc-tags-warn checkstyle:check).
run: |
if [ "${{ matrix.search-engine }}" = "opensearch" ]; then
- ./build.sh --ci --integration-tests --use-opensearch
+ ./build.sh --ci --integration-tests --skip-unit-tests --no-javadoc
--use-opensearch
else
- ./build.sh --ci --integration-tests
+ ./build.sh --ci --integration-tests --skip-unit-tests --no-javadoc
fi
# Keep only third-party dependencies in the post-job Maven cache:
Unomi's own
# snapshots are rebuilt every run and would only bloat the cache / risk
staleness
diff --git a/api/src/main/java/org/apache/unomi/api/tasks/ScheduledTask.java
b/api/src/main/java/org/apache/unomi/api/tasks/ScheduledTask.java
index c5d698e77..5a2f52ace 100644
--- a/api/src/main/java/org/apache/unomi/api/tasks/ScheduledTask.java
+++ b/api/src/main/java/org/apache/unomi/api/tasks/ScheduledTask.java
@@ -16,6 +16,7 @@
*/
package org.apache.unomi.api.tasks;
+import com.fasterxml.jackson.annotation.JsonIgnoreProperties;
import org.apache.unomi.api.Item;
import java.io.Serializable;
@@ -40,6 +41,11 @@ import java.util.HashSet;
* @see org.apache.unomi.api.services.SchedulerService
* @see TaskExecutor
*/
+// Tolerate unknown properties so a node running THIS version can still
deserialize task
+// documents written by a NEWER version that has added fields (rolling upgrade
window).
+// Without this, Jackson's default rejects the first unrecognized field and
the older node
+// loses access to all scheduler state until it is upgraded.
+@JsonIgnoreProperties(ignoreUnknown = true)
public class ScheduledTask extends Item implements Serializable {
/**
@@ -86,6 +92,7 @@ public class ScheduledTask extends Item implements
Serializable {
private boolean enabled;
private String lockOwner;
private Date lockDate;
+ private long lockLeaseMillis;
private boolean oneShot;
private boolean allowParallelExecution;
private TaskStatus status;
@@ -343,13 +350,40 @@ public class ScheduledTask extends Item implements
Serializable {
/**
* Sets the date when the current lock was acquired.
- *
+ *
* @param lockDate the lock acquisition date
*/
public void setLockDate(Date lockDate) {
this.lockDate = lockDate;
}
+ /**
+ * Duration in milliseconds for which the current lock is valid, as
declared by the node that
+ * acquired or last renewed it.
+ * <p>
+ * A lock's lifetime is a lease granted by its <em>owner</em>: the owner
renews it on a cadence
+ * derived from its own configured lock timeout, so only the owner's
timeout describes when a
+ * missing renewal actually means the owner is dead. Observers must judge
expiry against this
+ * recorded lease, never against their own configured timeout — a node
configured with a shorter
+ * timeout than the owner's renewal cadence would otherwise "recover" a
lock whose owner is alive
+ * and mid-execution, and the task would run twice.
+ *
+ * @return the lease duration in milliseconds, or {@code 0} when the lock
predates lease
+ * recording (legacy documents) and the observer's own timeout is
the only guide
+ */
+ public long getLockLeaseMillis() {
+ return lockLeaseMillis;
+ }
+
+ /**
+ * Sets the lease duration granted with the current lock.
+ *
+ * @param lockLeaseMillis the lease duration in milliseconds, {@code 0}
when unlocked or unknown
+ */
+ public void setLockLeaseMillis(long lockLeaseMillis) {
+ this.lockLeaseMillis = lockLeaseMillis;
+ }
+
/**
* Determines whether this task should execute only once.
* Tasks with period=0 are automatically marked as one-shot tasks.
diff --git a/build.sh b/build.sh
index c74fd5eb8..408768e32 100755
--- a/build.sh
+++ b/build.sh
@@ -279,8 +279,11 @@ RESOLVER_DEBUG=false
KEEP_CONTAINER=false
IT_SEARCH_ENGINE_LOGS=false
IT_MEMORY_SAMPLER=true
-IT_MEMORY_INTERVAL=30
+# 10s, matching itests/sample-it-memory.sh: at 30s a single sample spans
several ITs, so a
+# stall cannot be attributed to the test that caused it.
+IT_MEMORY_INTERVAL=10
JAVADOC=false
+NO_JAVADOC=false
LOG_FILE=""
LOG_FILE_ONLY=false
@@ -327,8 +330,9 @@ EOF
echo -e " ${CYAN}--keep-container${NC} Keep search engine
container running after tests (for post-failure inspection)"
echo -e " ${CYAN}--search-engine-logs${NC} Stream search engine
Docker logs to the Maven console during integration tests"
echo -e " ${CYAN}--no-memory-sampler${NC} Disable JVM/system
memory sampling during integration tests"
- echo -e " ${CYAN}--memory-interval SEC${NC} Memory sample interval
in seconds (default: 30)"
+ echo -e " ${CYAN}--memory-interval SEC${NC} Memory sample interval
in seconds (default: 10)"
echo -e " ${CYAN}--javadoc${NC} Build and validate
Javadoc after install (doclint errors fail; public/protected tag gaps warn)"
+ echo -e " ${CYAN}--no-javadoc${NC} Skip
Javadoc/checkstyle validation (overrides --ci; use when another job already ran
it)"
echo -e " ${CYAN}--ci${NC} CI mode: no Karaf,
non-interactive, includes Javadoc"
echo -e " ${CYAN}--log-file PATH${NC} Tee all output to
PATH (console + file)"
echo -e " ${CYAN}--log-file-only${NC} With --log-file:
write to file only, suppress console"
@@ -371,8 +375,9 @@ EOF
echo " --keep-container Keep search engine container running
after tests (for post-failure inspection)"
echo " --search-engine-logs Stream search engine Docker logs to
the Maven console during integration tests"
echo " --no-memory-sampler Disable JVM/system memory sampling
during integration tests"
- echo " --memory-interval SEC Memory sample interval in seconds
(default: 30)"
+ echo " --memory-interval SEC Memory sample interval in seconds
(default: 10)"
echo " --javadoc Build and validate Javadoc after
install (doclint errors fail; public/protected tag gaps warn)"
+ echo " --no-javadoc Skip Javadoc/checkstyle validation
(overrides --ci; use when another job already ran it)"
echo " --ci CI mode: no Karaf, non-interactive,
includes Javadoc"
echo " --log-file PATH Tee all output to PATH (console +
file)"
echo " --log-file-only With --log-file: write to file only,
suppress console"
@@ -549,6 +554,11 @@ while [ "$1" != "" ]; do
--javadoc)
JAVADOC=true
;;
+ --no-javadoc)
+ # Explicit veto, applied after argument parsing so it wins
regardless of whether it
+ # appears before or after --ci (which turns Javadoc on).
+ NO_JAVADOC=true
+ ;;
--log-file)
shift
LOG_FILE="$1"
@@ -1167,6 +1177,12 @@ echo "Estimated time: 3-5 minutes for build, 50-60
minutes with integration test
start_timer
# Build phases with enhanced output
+# Apply the --no-javadoc veto now that all arguments are parsed, so it wins
over --ci
+# regardless of flag order.
+if [ "$NO_JAVADOC" = true ]; then
+ JAVADOC=false
+fi
+
[ "$JAVADOC" = true ] && total_steps=4 || total_steps=2
current_step=0
diff --git a/itests/lib/it-run-memory.sh b/itests/lib/it-run-memory.sh
index 1c8f43466..8c61cdd5b 100644
--- a/itests/lib/it-run-memory.sh
+++ b/itests/lib/it-run-memory.sh
@@ -27,7 +27,7 @@ IT_MEMORY_SAMPLER_LOG="memory-sampler.log"
IT_MEMORY_SAMPLER_CACHE="memory-sampler.cache"
IT_MEMORY_SWAP_PRESSURE_MB=2048
-IT_MEMORY_TSV_HEADER=$'timestamp_utc\tkaraf_pid\tkaraf_heap_used_mb\tkaraf_heap_max_mb\tkaraf_gct_s\tes_heap_used_mb\tes_heap_max_mb\tdocker_rss_mb\tsystem_mem_available_mb\tsystem_swap_used_mb\tsystem_load_1m'
+IT_MEMORY_TSV_HEADER=$'timestamp_utc\tkaraf_pid\tkaraf_heap_used_mb\tkaraf_heap_max_mb\tkaraf_gct_s\tes_heap_used_mb\tes_heap_max_mb\tdocker_rss_mb\tsystem_mem_available_mb\tsystem_swap_used_mb\tsystem_load_1m\tkaraf_cpu_pct\tkaraf_io_read_mb_s\tkaraf_io_write_mb_s\tsearch_cpu_pct\tsearch_io_read_mb_s\tsearch_io_write_mb_s'
_IT_MEMORY_OS=""
@@ -350,7 +350,12 @@ it_memory_parse_docker_mem_to_mb() {
}
it_memory_find_karaf_pid() {
- pgrep -f 'org.apache.karaf.main.Main' 2>/dev/null | head -1
+ # `|| true`: pgrep exits non-zero when nothing matches, and with `set -euo
pipefail` that
+ # aborted the whole sample. The sampler starts before Karaf does, so every
sample taken
+ # during startup was discarded -- exactly the window where the search
engine is booting and
+ # its resource use is most interesting. No match now yields an empty pid,
which the callers
+ # and the summarizer already treat as "no Karaf yet" (guarded by `if ($2+0
> 0)`).
+ pgrep -f 'org.apache.karaf.main.Main' 2>/dev/null | head -1 || true
}
it_memory_karaf_max_mb_cached() {
@@ -446,21 +451,139 @@ it_memory_search_engine_stats() {
echo -e "$(it_memory_mb_from_bytes
"${used_bytes:-0}")\t$(it_memory_mb_from_bytes "${max_bytes:-0}")"
}
-it_memory_docker_rss_mb() {
+# --- CPU and disk I/O sampling
-------------------------------------------------
+#
+# Added to answer "is the run CPU-bound, I/O-bound, or waiting?". The memory
columns alone
+# could not distinguish a busy run from an idle one blocked on a remote call,
which is exactly
+# the question raised by the Elasticsearch/OpenSearch IT duration gap: system
load was near
+# idle on the slower engine, so the extra time was spent waiting rather than
computing.
+#
+# CPU is measured as a TRUE INTERVAL PERCENTAGE, not ps(1)'s %cpu -- that is
an average over the
+# whole process lifetime, so a JVM that was busy at startup reads as busy
forever and the number
+# is useless for spotting a stall. On Linux -- which is what CI runs, and the
only place these
+# numbers are compared across runs -- we delta /proc/<pid>/stat between
samples for a true
+# interval figure. macOS has no procfs, so it falls back to ps(1)'s lifetime
average: good enough
+# to see that a process is alive and roughly how hard it has worked, but NOT
comparable with a
+# Linux sample and not to be read as "CPU right now". Everything here is
best-effort: a missing
+# file, a dead pid or an absent docker CLI yields 0 and never fails a run.
+
+# Stores "value timestamp" pairs so the next sample can compute a delta.
+_it_memory_counter_cache() {
+ local target_dir="$1" key="$2"
+ echo "$target_dir/.it-memory-counter-$key"
+}
+
+# Echoes the per-second rate between this reading and the previous one, or 0
on the first call.
+_it_memory_rate_per_sec() {
+ local target_dir="$1" key="$2" value="$3"
+ local cache prev_value prev_ts now delta_v delta_t
+ cache="$(_it_memory_counter_cache "$target_dir" "$key")"
+ now="$(date +%s)"
+
+ if [ -r "$cache" ]; then
+ read -r prev_value prev_ts < "$cache" 2>/dev/null || true
+ fi
+ printf '%s %s\n' "$value" "$now" > "$cache" 2>/dev/null || true
+
+ if [ -z "${prev_value:-}" ] || [ -z "${prev_ts:-}" ]; then
+ echo "0"
+ return
+ fi
+ delta_t=$((now - prev_ts))
+ [ "$delta_t" -le 0 ] && { echo "0"; return; }
+ delta_v="$(awk -v a="$value" -v b="$prev_value" 'BEGIN { d = a - b; print
(d > 0 ? d : 0) }')"
+ awk -v d="$delta_v" -v t="$delta_t" 'BEGIN { printf "%.2f", d / t }'
+}
+
+# Interval CPU% for a pid. >100 is legitimate on multi-core (sum across
threads).
+it_memory_process_cpu_pct() {
+ local target_dir="$1" pid="${2:-}"
+ local ticks hz cpu_s rate
+
+ if [ -z "$pid" ] || [ "$pid" = "0" ] || ! kill -0 "$pid" 2>/dev/null; then
+ echo "0"
+ return
+ fi
+
+ if it_memory_is_linux && [ -r "/proc/$pid/stat" ]; then
+ # Fields 14 (utime) and 15 (stime), in clock ticks. comm (field 2) is
parenthesised and
+ # may itself contain spaces AND parentheses, so split after the LAST
')' rather than the
+ # first: a process named e.g. "java (worker)" otherwise shifts every
subsequent index.
+ ticks="$(awk '{
+ i = length($0)
+ while (i > 0 && substr($0, i, 1) != ")") i--
+ n = split(substr($0, i + 2), f, " ")
+ if (n >= 13) print f[12] + f[13]; else print 0
+ }' "/proc/$pid/stat" 2>/dev/null)"
+ [ -z "$ticks" ] && { echo "0"; return; }
+ hz="$(getconf CLK_TCK 2>/dev/null || echo 100)"
+ cpu_s="$(awk -v t="$ticks" -v hz="$hz" 'BEGIN { printf "%.4f", t / hz
}')"
+ rate="$(_it_memory_rate_per_sec "$target_dir" "cpu-$pid" "$cpu_s")"
+ awk -v r="$rate" 'BEGIN { printf "%.1f", r * 100 }'
+ return
+ fi
+
+ # macOS / no procfs: lifetime average, better than nothing for a local run.
+ ps -o %cpu= -p "$pid" 2>/dev/null | tr -d ' ' | awk 'NF { printf "%.1f",
$1; found = 1 } END { if (!found) print 0 }'
+}
+
+# Interval disk read/write in MB/s for a pid (Linux only; /proc/<pid>/io).
+it_memory_process_io_mb_s() {
+ local target_dir="$1" pid="${2:-}"
+ local read_bytes write_bytes read_rate write_rate
+
+ if [ -z "$pid" ] || [ "$pid" = "0" ] || ! it_memory_is_linux || [ ! -r
"/proc/$pid/io" ]; then
+ echo -e "0\t0"
+ return
+ fi
+
+ read_bytes="$(awk '/^read_bytes:/ { print $2 }' "/proc/$pid/io"
2>/dev/null)"
+ write_bytes="$(awk '/^write_bytes:/ { print $2 }' "/proc/$pid/io"
2>/dev/null)"
+ read_rate="$(_it_memory_rate_per_sec "$target_dir" "ior-$pid"
"${read_bytes:-0}")"
+ write_rate="$(_it_memory_rate_per_sec "$target_dir" "iow-$pid"
"${write_bytes:-0}")"
+ awk -v r="$read_rate" -v w="$write_rate" 'BEGIN { printf "%.2f\t%.2f", r /
1048576, w / 1048576 }'
+}
+
+# One docker stats call per sample, returning RSS, CPU% and block I/O together.
+#
+# Deliberately a single invocation: `docker stats --no-stream` costs ~1s and
briefly loads the
+# daemon, and the sampler now runs 3x more often (10s rather than 30s). Two
calls per sample
+# would have meant six times the docker traffic of the original sampler,
perturbing the very
+# run being measured and tripling the exposure to a hung daemon. CPUPerc is
already an interval
+# measurement; BlockIO is cumulative and is deltaed here.
+#
+# Echoes: rss_mb \t cpu_pct \t io_read_mb_s \t io_write_mb_s
+it_memory_docker_sample() {
local target_dir="$1"
- local container rss
+ local container stats mem cpu blockio read_raw write_raw read_b write_b
read_rate write_rate
if ! command -v docker >/dev/null 2>&1; then
- echo "0"
+ echo -e "0\t0\t0\t0"
return
fi
container="$(it_memory_resolve_docker_container "$target_dir")"
- rss="$(docker stats --no-stream --format '{{.MemUsage}}' "$container"
2>/dev/null | head -1 | cut -d/ -f1 | tr -d ' ')"
+ stats="$(docker stats --no-stream --format
'{{.MemUsage}}|{{.CPUPerc}}|{{.BlockIO}}' "$container" 2>/dev/null | head -1)"
+ if [ -z "$stats" ]; then
+ echo -e "0\t0\t0\t0"
+ return
+ fi
+
+ mem="$(echo "$stats" | cut -d'|' -f1 | cut -d/ -f1 | tr -d ' ')"
+ cpu="$(echo "$stats" | cut -d'|' -f2 | tr -d ' %')"
+ blockio="$(echo "$stats" | cut -d'|' -f3)"
+ read_raw="$(echo "$blockio" | cut -d/ -f1 | tr -d ' ')"
+ write_raw="$(echo "$blockio" | cut -d/ -f2 | tr -d ' ')"
+ read_b="$(it_memory_parse_docker_mem_to_mb "$read_raw")"
+ write_b="$(it_memory_parse_docker_mem_to_mb "$write_raw")"
+ read_rate="$(_it_memory_rate_per_sec "$target_dir" "dior" "${read_b:-0}")"
+ write_rate="$(_it_memory_rate_per_sec "$target_dir" "diow"
"${write_b:-0}")"
- it_memory_parse_docker_mem_to_mb "$rss"
+ printf '%s\t%.1f\t%.2f\t%.2f\n' \
+ "$(it_memory_parse_docker_mem_to_mb "$mem")" "${cpu:-0}"
"${read_rate:-0}" "${write_rate:-0}"
}
+
it_memory_system_stats() {
local mem_available swap_used load_1m
@@ -482,13 +605,22 @@ it_memory_sample_once() {
es_line="$(it_memory_search_engine_stats "$port")"
sys_line="$(it_memory_system_stats)"
- printf '%s\t%s\t%s\t%s\t%s\t%s\n' \
+ # One docker call per sample; split into the RSS column (8) and the CPU/IO
columns (15-17).
+ local docker_line docker_rss docker_cpu_io
+ docker_line="$(it_memory_docker_sample "$target_dir")"
+ docker_rss="$(printf '%s' "$docker_line" | cut -f1)"
+ docker_cpu_io="$(printf '%s' "$docker_line" | cut -f2-4)"
+
+ printf '%s\t%s\t%s\t%s\t%s\t%s\t%s\t%s\t%s\n' \
"$(date -u +%Y-%m-%dT%H:%M:%SZ)" \
"${karaf_pid:-0}" \
"$karaf_line" \
"$es_line" \
- "$(it_memory_docker_rss_mb "$target_dir")" \
- "$sys_line"
+ "${docker_rss:-0}" \
+ "$sys_line" \
+ "$(it_memory_process_cpu_pct "$target_dir" "$karaf_pid")" \
+ "$(it_memory_process_io_mb_s "$target_dir" "$karaf_pid")" \
+ "${docker_cpu_io:-$(printf '0\t0\t0')}"
}
it_memory_write_samples_header() {
@@ -508,7 +640,7 @@ it_memory_summarize_samples() {
awk -F'\t' -v summary="$summary_file" -v
swap_pressure_mb="$IT_MEMORY_SWAP_PRESSURE_MB" '
NR == 1 { next }
- NF < 11 { next }
+ NF < 11 { next } # pre-CPU/IO samples still summarize
{
samples++
if ($2+0 > 0) {
@@ -524,6 +656,18 @@ it_memory_summarize_samples() {
if ($11+0 > peak_load) peak_load = $11+0
if (samples == 1) first_swap = $10+0
last_swap = $10+0
+ # CPU / IO columns are absent in samples written before they were
added.
+ if (NF >= 17) {
+ cpu_samples++
+ karaf_cpu_sum += $12+0; if ($12+0 > peak_karaf_cpu)
peak_karaf_cpu = $12+0
+ search_cpu_sum += $15+0; if ($15+0 > peak_search_cpu)
peak_search_cpu = $15+0
+ io_sum += $13+0 + $14+0 + $16+0 + $17+0
+ if ($13+0 + $14+0 > peak_karaf_io) peak_karaf_io = $13+0 +
$14+0
+ if ($16+0 + $17+0 > peak_search_io) peak_search_io = $16+0 +
$17+0
+ # "Idle" = neither process using meaningful CPU: the signature
of a run that is
+ # waiting on latency rather than doing work.
+ if ($12+0 < 10 && $15+0 < 10) idle_samples++
+ }
}
END {
if (samples == 0) exit 1
@@ -541,6 +685,24 @@ it_memory_summarize_samples() {
printf("memory.min.system.mem.available.mb=%d\n", min_mem_avail+0)
>> summary
printf("memory.peak.system.swap.used.mb=%d\n", peak_swap+0) >>
summary
printf("memory.peak.system.load.1m=%.2f\n", peak_load+0) >> summary
+ if (cpu_samples > 0) {
+ printf("cpu.samples.count=%d\n", cpu_samples) >> summary
+ printf("cpu.mean.karaf.pct=%.1f\n", karaf_cpu_sum /
cpu_samples) >> summary
+ printf("cpu.peak.karaf.pct=%.1f\n", peak_karaf_cpu+0) >>
summary
+ printf("cpu.mean.search.pct=%.1f\n", search_cpu_sum /
cpu_samples) >> summary
+ printf("cpu.peak.search.pct=%.1f\n", peak_search_cpu+0) >>
summary
+ printf("cpu.idle.samples.pct=%d\n", idle_samples * 100 /
cpu_samples) >> summary
+ printf("io.peak.karaf.mb.s=%.2f\n", peak_karaf_io+0) >> summary
+ printf("io.peak.search.mb.s=%.2f\n", peak_search_io+0) >>
summary
+ printf("io.mean.total.mb.s=%.2f\n", io_sum / cpu_samples) >>
summary
+ # Mostly-idle CPU with negligible I/O means the run is
latency-bound: time is
+ # going on waiting (polls, refresh intervals, timeouts), not
on work.
+ if (idle_samples * 100 / cpu_samples >= 70 && io_sum /
cpu_samples < 5) {
+ printf("cpu.warning.mostly.idle=true\n") >> summary
+ } else {
+ printf("cpu.warning.mostly.idle=false\n") >> summary
+ }
+ }
printf("memory.karaf.headroom.pct=%d\n", karaf_headroom+0) >>
summary
printf("memory.search.headroom.pct=%d\n", es_headroom+0) >> summary
if (swap_pressure) {
diff --git a/itests/sample-it-memory.sh b/itests/sample-it-memory.sh
index 19e0d5d5c..e0469bd02 100755
--- a/itests/sample-it-memory.sh
+++ b/itests/sample-it-memory.sh
@@ -39,7 +39,10 @@ source "$SCRIPT_DIR/lib/it-run.sh"
source "$SCRIPT_DIR/lib/it-run-memory.sh"
TARGET_DIR="$SCRIPT_DIR/target"
-INTERVAL=30
+# 10s, down from 30s: at 30s a sample covers several ITs at once, so a stall
cannot be
+# attributed to the test that caused it. Each sample is a few cheap reads plus
one
+# `docker stats --no-stream`, and a 50-minute run produces ~300 rows (a few
tens of KB).
+INTERVAL=10
SEARCH_PORT=""
PRINT_ONLY=false
COMMAND=""
@@ -58,7 +61,7 @@ Commands:
Options:
--target-dir DIR IT target directory (default: itests/target)
- --interval SEC Sample interval in seconds for start (default: 30)
+ --interval SEC Sample interval in seconds for start (default: 10)
--port PORT Search engine HTTP port override
--print-only With operator-note: print to stdout instead of writing
file
-h, --help Show this help
@@ -78,7 +81,7 @@ parse_args() {
;;
--interval)
shift
- INTERVAL="${1:-30}"
+ INTERVAL="${1:-10}"
;;
--port)
shift
diff --git
a/persistence-elasticsearch/core/src/main/resources/META-INF/cxs/mappings/scheduledTask.json
b/persistence-elasticsearch/core/src/main/resources/META-INF/cxs/mappings/scheduledTask.json
index f36fc297c..030305e8e 100644
---
a/persistence-elasticsearch/core/src/main/resources/META-INF/cxs/mappings/scheduledTask.json
+++
b/persistence-elasticsearch/core/src/main/resources/META-INF/cxs/mappings/scheduledTask.json
@@ -75,6 +75,9 @@
"lockDate": {
"type": "date"
},
+ "lockLeaseMillis": {
+ "type": "long"
+ },
"lastExecutionDate": {
"type": "date"
},
diff --git
a/persistence-opensearch/core/src/main/resources/META-INF/cxs/mappings/scheduledTask.json
b/persistence-opensearch/core/src/main/resources/META-INF/cxs/mappings/scheduledTask.json
index 9c1541d96..a251eebf4 100644
---
a/persistence-opensearch/core/src/main/resources/META-INF/cxs/mappings/scheduledTask.json
+++
b/persistence-opensearch/core/src/main/resources/META-INF/cxs/mappings/scheduledTask.json
@@ -78,6 +78,9 @@
"lockDate": {
"type": "date"
},
+ "lockLeaseMillis": {
+ "type": "long"
+ },
"lastExecutionDate": {
"type": "date"
},
diff --git
a/services/src/main/java/org/apache/unomi/services/impl/scheduler/SchedulerServiceImpl.java
b/services/src/main/java/org/apache/unomi/services/impl/scheduler/SchedulerServiceImpl.java
index 3c98963fc..5e84267da 100644
---
a/services/src/main/java/org/apache/unomi/services/impl/scheduler/SchedulerServiceImpl.java
+++
b/services/src/main/java/org/apache/unomi/services/impl/scheduler/SchedulerServiceImpl.java
@@ -493,6 +493,7 @@ public class SchedulerServiceImpl implements
SchedulerService {
if (newStatus == TaskStatus.COMPLETED || newStatus ==
TaskStatus.FAILED) {
task.setLockOwner(null);
task.setLockDate(null);
+ task.setLockLeaseMillis(0);
task.setWaitingForTaskType(null);
task.setCurrentStep(null);
// Update last execution date for completed/failed tasks
@@ -511,6 +512,7 @@ public class SchedulerServiceImpl implements
SchedulerService {
} else if (newStatus == TaskStatus.WAITING) {
task.setLockOwner(null);
task.setLockDate(null);
+ task.setLockLeaseMillis(0);
} else if (newStatus == TaskStatus.RUNNING) {
// Update status details for running tasks
Map<String, Object> details = task.getStatusDetails();
@@ -899,6 +901,7 @@ public class SchedulerServiceImpl implements
SchedulerService {
// and PersistenceSchedulerProvider.preDestroy
need not unlock RUNNING.
task.setLockOwner(null);
task.setLockDate(null);
+ task.setLockLeaseMillis(0);
if (task.isPersistent() && persistenceProvider !=
null) {
if (!persistenceProvider.saveTask(task)) {
LOGGER.warn("Failed to persist CRASHED
state for task {} during shutdown; "
diff --git
a/services/src/main/java/org/apache/unomi/services/impl/scheduler/TaskExecutionManager.java
b/services/src/main/java/org/apache/unomi/services/impl/scheduler/TaskExecutionManager.java
index 8f3bd41a4..98b98a2e3 100644
---
a/services/src/main/java/org/apache/unomi/services/impl/scheduler/TaskExecutionManager.java
+++
b/services/src/main/java/org/apache/unomi/services/impl/scheduler/TaskExecutionManager.java
@@ -23,6 +23,8 @@ import org.slf4j.LoggerFactory;
import java.util.ArrayList;
import java.util.Date;
+import java.util.HashMap;
+import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.*;
@@ -502,6 +504,7 @@ public class TaskExecutionManager {
task.setExecutingNodeId(null);
task.setLockOwner(null);
task.setLockDate(null);
+ task.setLockLeaseMillis(0);
schedulerService.saveTask(task, true);
} catch (Exception e) {
LOGGER.warn("Failed to abort prepared task {} during shutdown: {}",
@@ -532,6 +535,16 @@ public class TaskExecutionManager {
return;
}
long interval = Math.max(MIN_LOCK_RENEWAL_INTERVAL_MS,
lockManager.getLockTimeout() / 3);
+ if (interval >= lockManager.getLockTimeout()) {
+ // The renewal floor exceeds the configured timeout, so this node
cannot renew its own
+ // lease fast enough to keep it alive: peers may legitimately
treat its live locks as
+ // expired between two renewals and recover mid-execution tasks.
Surface the
+ // misconfiguration instead of leaving sporadic double executions
to be diagnosed.
+ LOGGER.warn("Lock timeout {}ms is at or below the minimum renewal
interval {}ms: "
+ + "this node's live locks can expire between renewals and
be recovered by peers. "
+ + "Configure a lock timeout of at least {}ms.",
+ lockManager.getLockTimeout(), interval,
MIN_LOCK_RENEWAL_INTERVAL_MS * 3);
+ }
LockRenewalHandle handle = new LockRenewalHandle();
activeLockRenewals.put(task.getItemId(), handle);
try {
@@ -657,9 +670,57 @@ public class TaskExecutionManager {
// Carry OCC tokens from the fresh load so persistTerminalState can
CAS.
TaskLockManager.copyOccMetadata(latest, task);
+ rebaseAccumulatorsFromStore(task, latest);
return true;
}
+ /**
+ * Rebases the running task's accumulating fields on the authoritative
store document before a
+ * terminal handler increments them.
+ * <p>
+ * The task instance a wrapper carries comes from the dispatch path, whose
discovery query
+ * ({@code findEnabledScheduledOrWaitingTasks}) is search-based and
therefore lags the store by
+ * up to the index refresh interval. Its {@code successCount}, {@code
failureCount} and
+ * execution history can predate writes that have already landed. The
compare-and-set in
+ * {@link #persistTerminalState} protects only the document
<em>version</em>, not these values:
+ * incrementing a stale base and then CAS-writing it succeeds and silently
loses the newer
+ * count. Observed as a periodic task reporting one success after two
successful executions.
+ * <p>
+ * Only accumulators are taken from the store. Status, lock fields and
scheduling are the
+ * terminal handler's business and are set from the execution's own
outcome.
+ *
+ * @param task the executing task instance about to be mutated by a
terminal handler
+ * @param latest the authoritative document, freshly loaded by id
+ */
+ private static void rebaseAccumulatorsFromStore(ScheduledTask task,
ScheduledTask latest) {
+ task.setSuccessCount(latest.getSuccessCount());
+ task.setFailureCount(latest.getFailureCount());
+
+ // Execution history is append-only, so the longer list is the more
current one. Other
+ // statusDetails keys stay as the execution left them (checkpoint
markers, crash details).
+ Map<String, Object> latestDetails = latest.getStatusDetails();
+ if (latestDetails == null) {
+ return;
+ }
+ Object latestHistory = latestDetails.get("executionHistory");
+ if (!(latestHistory instanceof List)) {
+ return;
+ }
+ Map<String, Object> details = task.getStatusDetails();
+ if (details == null) {
+ details = new HashMap<>();
+ task.setStatusDetails(details);
+ } else if (!(details instanceof HashMap)) {
+ details = new HashMap<>(details);
+ task.setStatusDetails(details);
+ }
+ Object ourHistory = details.get("executionHistory");
+ int ourSize = ourHistory instanceof List ? ((List<?>)
ourHistory).size() : 0;
+ if (((List<?>) latestHistory).size() > ourSize) {
+ details.put("executionHistory", new ArrayList<>((List<?>)
latestHistory));
+ }
+ }
+
/**
* Persists a terminal task state. Persistent tasks use compare-and-set so
a late
* complete/fail cannot clobber CANCELLED or a peer's RUNNING document.
Lock fields are
@@ -668,6 +729,7 @@ public class TaskExecutionManager {
private boolean persistTerminalState(ScheduledTask task) {
task.setLockOwner(null);
task.setLockDate(null);
+ task.setLockLeaseMillis(0);
if (!task.isPersistent()) {
boolean saved = schedulerService.saveTask(task);
if (!saved) {
diff --git
a/services/src/main/java/org/apache/unomi/services/impl/scheduler/TaskLockManager.java
b/services/src/main/java/org/apache/unomi/services/impl/scheduler/TaskLockManager.java
index 978d97fa6..fbe1517c0 100644
---
a/services/src/main/java/org/apache/unomi/services/impl/scheduler/TaskLockManager.java
+++
b/services/src/main/java/org/apache/unomi/services/impl/scheduler/TaskLockManager.java
@@ -161,6 +161,7 @@ public class TaskLockManager {
// Just set lock info but don't enforce exclusivity
task.setLockOwner(nodeId);
task.setLockDate(new Date());
+ task.setLockLeaseMillis(lockTimeout);
metricsManager.updateMetric(TaskMetricsManager.METRIC_TASKS_LOCK_ACQUIRED);
return true;
}
@@ -194,8 +195,10 @@ public class TaskLockManager {
latest.setLockOwner(nodeId);
latest.setLockDate(new Date());
+ latest.setLockLeaseMillis(lockTimeout);
task.setLockOwner(nodeId);
task.setLockDate(latest.getLockDate());
+ task.setLockLeaseMillis(lockTimeout);
metricsManager.updateMetric(TaskMetricsManager.METRIC_TASKS_LOCK_ACQUIRED);
// For non-persistent tasks, we just update the in-memory map
@@ -247,9 +250,12 @@ public class TaskLockManager {
task.setSystemMetadata(SEQ_NO, latestTask.getSystemMetadata(SEQ_NO));
task.setSystemMetadata(PRIMARY_TERM,
latestTask.getSystemMetadata(PRIMARY_TERM));
- // Step 6: Set lock information
+ // Step 6: Set lock information. The lease records THIS node's timeout
with the lock:
+ // renewal cadence is derived from the owner's timeout, so only the
owner's timeout says
+ // when a missing renewal means the owner is dead (see
isLockExpired()).
task.setLockOwner(nodeId);
task.setLockDate(new Date());
+ task.setLockLeaseMillis(lockTimeout);
LOGGER.debug("LOCK-DIAG [{}] node {} : attempting CAS write -
if_seq_no={}, if_primary_term={}, "
+ "writing lockOwner={}",
@@ -391,6 +397,7 @@ public class TaskLockManager {
if (latestOwner == null) {
task.setLockOwner(null);
task.setLockDate(null);
+ task.setLockLeaseMillis(0);
LOGGER.debug("LOCK-DIAG [{}] node {} : releaseLock() no-op,
store already unlocked",
task.getItemId(), nodeId);
return true;
@@ -406,8 +413,10 @@ public class TaskLockManager {
toSave.setLockOwner(null);
toSave.setLockDate(null);
+ toSave.setLockLeaseMillis(0);
task.setLockOwner(null);
task.setLockDate(null);
+ task.setLockLeaseMillis(0);
// Compare-and-set on the freshly loaded seq_no/primary_term, not
a blind overwrite:
// a peer may win a legitimate CAS-based lock acquisition in the
window between our
@@ -474,6 +483,7 @@ public class TaskLockManager {
}
latest.setLockDate(new Date());
+ latest.setLockLeaseMillis(lockTimeout);
// Compare-and-set on the fresh store view: if a peer stole the
lock between the
// read above and this write, renewal fails closed instead of
resurrecting our lock.
@@ -486,6 +496,7 @@ public class TaskLockManager {
// the executing thread's later compare-and-set writes are checked
against the
// store's current version, not the pre-renewal one.
task.setLockDate(latest.getLockDate());
+ task.setLockLeaseMillis(latest.getLockLeaseMillis());
copyOccMetadata(latest, task);
LOGGER.debug("LOCK-DIAG [{}] node {} : renewLock() succeeded, new
lockDate={}",
task.getItemId(), nodeId, latest.getLockDate());
@@ -537,12 +548,22 @@ public class TaskLockManager {
return true;
}
+ // Judge expiry against the lease the OWNER recorded with the lock,
not this node's own
+ // configured timeout. The owner renews on a cadence derived from its
own timeout
+ // (lockTimeout/3, see TaskExecutionManager#startLockRenewal), so a
node configured with a
+ // shorter timeout than the owner's renewal cadence would otherwise
declare a live,
+ // renewed lock dead in the gap between two renewals and "recover" a
task that is still
+ // executing — observed as double execution under divergent per-node
configuration.
+ // Locks written before lease recording carry no lease (0); only for
those does this
+ // node's own timeout remain the best available guess.
+ long lease = task.getLockLeaseMillis() > 0 ? task.getLockLeaseMillis()
: lockTimeout;
long now = System.currentTimeMillis();
long lockAge = now - task.getLockDate().getTime();
- boolean expired = lockAge > lockTimeout;
+ boolean expired = lockAge > lease;
LOGGER.debug("LOCK-DIAG isLockExpired() : task={}, lockDate={} ({}),
now={}, lockAge={}ms, "
- + "lockTimeout={}ms -> expired={}",
- task.getItemId(), task.getLockDate(),
task.getLockDate().getTime(), now, lockAge, lockTimeout, expired);
+ + "lease={}ms (recorded={}ms, own timeout={}ms) -> expired={}",
+ task.getItemId(), task.getLockDate(),
task.getLockDate().getTime(), now, lockAge,
+ lease, task.getLockLeaseMillis(), lockTimeout, expired);
return expired;
}
}
diff --git
a/services/src/main/java/org/apache/unomi/services/impl/scheduler/TaskRecoveryManager.java
b/services/src/main/java/org/apache/unomi/services/impl/scheduler/TaskRecoveryManager.java
index 43b29e368..d41c34c30 100644
---
a/services/src/main/java/org/apache/unomi/services/impl/scheduler/TaskRecoveryManager.java
+++
b/services/src/main/java/org/apache/unomi/services/impl/scheduler/TaskRecoveryManager.java
@@ -255,6 +255,7 @@ public class TaskRecoveryManager {
}
latest.setLockOwner(null);
latest.setLockDate(null);
+ latest.setLockLeaseMillis(0);
// Record the crash in execution history
recordCrash(latest, previousOwner);
@@ -273,6 +274,7 @@ public class TaskRecoveryManager {
task.setStatus(latest.getStatus());
task.setLockOwner(null);
task.setLockDate(null);
+ task.setLockLeaseMillis(0);
task.setStatusDetails(latest.getStatusDetails());
task.setCurrentStep(latest.getCurrentStep());
task.setLastError(latest.getLastError());
diff --git
a/services/src/main/java/org/apache/unomi/services/impl/scheduler/TaskStateManager.java
b/services/src/main/java/org/apache/unomi/services/impl/scheduler/TaskStateManager.java
index c8877bd72..d8bf91266 100644
---
a/services/src/main/java/org/apache/unomi/services/impl/scheduler/TaskStateManager.java
+++
b/services/src/main/java/org/apache/unomi/services/impl/scheduler/TaskStateManager.java
@@ -148,6 +148,7 @@ public class TaskStateManager {
private void clearTaskExecution(ScheduledTask task) {
task.setLockOwner(null);
task.setLockDate(null);
+ task.setLockLeaseMillis(0);
task.setWaitingForTaskType(null);
task.setCurrentStep(null);
}
@@ -162,6 +163,7 @@ public class TaskStateManager {
private void clearLockInfo(ScheduledTask task) {
task.setLockOwner(null);
task.setLockDate(null);
+ task.setLockLeaseMillis(0);
}
private void updateRunningState(ScheduledTask task, String nodeId) {
diff --git
a/services/src/test/java/org/apache/unomi/services/impl/scheduler/ScheduledTaskLeaseSerializationTest.java
b/services/src/test/java/org/apache/unomi/services/impl/scheduler/ScheduledTaskLeaseSerializationTest.java
new file mode 100644
index 000000000..126cee701
--- /dev/null
+++
b/services/src/test/java/org/apache/unomi/services/impl/scheduler/ScheduledTaskLeaseSerializationTest.java
@@ -0,0 +1,124 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.unomi.services.impl.scheduler;
+
+import org.apache.unomi.api.Item;
+import org.apache.unomi.api.tasks.ScheduledTask;
+import org.apache.unomi.persistence.spi.CustomObjectMapper;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+import java.util.Date;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+/**
+ * Persistence-format coverage for {@link ScheduledTask#getLockLeaseMillis()}.
+ * <p>
+ * The lock lease is a cross-node security decision (it decides who may
declare a peer dead), so
+ * its survival through the production serializer is not an implementation
detail: a field that
+ * silently fails to round-trip would degrade every observer to the legacy
observer-timeout
+ * fallback and quietly reintroduce the divergent-timeout double-execution
bug. Both store read
+ * paths are exercised: direct class binding, and the {@code Item}-dispatched
path the persistence
+ * services actually use ({@code readValue(json, Item.class)} via {@code
ItemDeserializer}).
+ */
+public class ScheduledTaskLeaseSerializationTest {
+
+ private CustomObjectMapper mapper;
+
+ @BeforeEach
+ public void setUp() {
+ mapper = CustomObjectMapper.getCustomInstance();
+ mapper.registerBuiltInItemTypeClass(ScheduledTask.ITEM_TYPE,
ScheduledTask.class);
+ }
+
+ private ScheduledTask lockedTask() {
+ ScheduledTask task = new ScheduledTask();
+ task.setItemId("lease-serialization-test");
+ task.setTaskType("lease-serialization-test");
+ task.setStatus(ScheduledTask.TaskStatus.RUNNING);
+ task.setLockOwner("node-a");
+ task.setLockDate(new Date());
+ task.setLockLeaseMillis(12345);
+ return task;
+ }
+
+ @Test
+ public void leaseSurvivesRoundTripViaDirectClassBinding() throws Exception
{
+ String json = mapper.writeValueAsString(lockedTask());
+ assertTrue(json.contains("\"lockLeaseMillis\":12345"), "lease must be
serialized: " + json);
+
+ ScheduledTask back = mapper.readValue(json, ScheduledTask.class);
+ assertEquals(12345, back.getLockLeaseMillis());
+ assertEquals("node-a", back.getLockOwner());
+ }
+
+ @Test
+ public void leaseSurvivesRoundTripViaItemDispatchedPath() throws Exception
{
+ // This is the path the persistence services use when loading store
documents.
+ String json = mapper.writeValueAsString(lockedTask());
+ Item item = mapper.readValue(json, Item.class);
+ assertTrue(item instanceof ScheduledTask, "itemType dispatch should
yield a ScheduledTask");
+ assertEquals(12345, ((ScheduledTask) item).getLockLeaseMillis());
+ }
+
+ /**
+ * A document written BEFORE lease recording (no {@code lockLeaseMillis}
field) must load with
+ * lease 0, which {@code TaskLockManager#isLockExpired} treats as "fall
back to the observer's
+ * own timeout" — i.e. exactly the pre-lease behaviour, so a rolling
upgrade cannot make old
+ * locks unexpirable or instantly expired.
+ */
+ @Test
+ public void legacyDocumentWithoutLeaseLoadsAsZero() throws Exception {
+ String legacyJson = "{" +
+ "\"itemId\":\"legacy-task\"," +
+ "\"itemType\":\"scheduledTask\"," +
+ "\"taskType\":\"legacy-task\"," +
+ "\"status\":\"RUNNING\"," +
+ "\"lockOwner\":\"old-node\"," +
+ "\"lockDate\":\"2026-01-01T00:00:00Z\"" +
+ "}";
+ Item item = mapper.readValue(legacyJson, Item.class);
+ ScheduledTask task = (ScheduledTask) item;
+ assertEquals(0, task.getLockLeaseMillis(), "missing lease must read as
0 (legacy fallback)");
+ assertEquals("old-node", task.getLockOwner());
+ }
+
+ /**
+ * A document written by a NEWER version carrying a field this version
does not know must
+ * still deserialize (rolling upgrade window: older binaries keep reading
scheduler state
+ * written by upgraded peers). Pinned by {@code
@JsonIgnoreProperties(ignoreUnknown = true)}
+ * on ScheduledTask — without it, Jackson's default rejects the first
unknown field and the
+ * older node loses access to every task document the newer node has
touched.
+ */
+ @Test
+ public void documentFromNewerVersionWithUnknownFieldStillLoads() throws
Exception {
+ String futureJson = "{" +
+ "\"itemId\":\"future-task\"," +
+ "\"itemType\":\"scheduledTask\"," +
+ "\"taskType\":\"future-task\"," +
+ "\"status\":\"SCHEDULED\"," +
+ "\"lockLeaseMillis\":5000," +
+ "\"someFieldAddedInAFutureVersion\":\"whatever\"" +
+ "}";
+ Item item = mapper.readValue(futureJson, Item.class);
+ assertNotNull(item);
+ assertEquals(5000, ((ScheduledTask) item).getLockLeaseMillis());
+ }
+}
diff --git
a/services/src/test/java/org/apache/unomi/services/impl/scheduler/SchedulerDiagnosticsExtension.java
b/services/src/test/java/org/apache/unomi/services/impl/scheduler/SchedulerDiagnosticsExtension.java
new file mode 100644
index 000000000..73bf1eb5d
--- /dev/null
+++
b/services/src/test/java/org/apache/unomi/services/impl/scheduler/SchedulerDiagnosticsExtension.java
@@ -0,0 +1,271 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.unomi.services.impl.scheduler;
+
+import ch.qos.logback.classic.Level;
+import ch.qos.logback.classic.LoggerContext;
+import ch.qos.logback.classic.spi.ILoggingEvent;
+import ch.qos.logback.core.read.CyclicBufferAppender;
+import org.apache.unomi.api.tasks.ScheduledTask;
+import org.apache.unomi.persistence.spi.PersistenceService;
+import org.junit.jupiter.api.extension.AfterEachCallback;
+import org.junit.jupiter.api.extension.BeforeEachCallback;
+import org.junit.jupiter.api.extension.ExtensionContext;
+import org.junit.jupiter.api.extension.TestExecutionExceptionHandler;
+import org.slf4j.LoggerFactory;
+
+import java.lang.reflect.Field;
+import java.text.SimpleDateFormat;
+import java.util.Collection;
+import java.util.Date;
+import java.util.List;
+import java.util.Map;
+
+/**
+ * Makes an intermittent scheduler test failure diagnosable from the CI log it
failed in.
+ * <p>
+ * Scheduler failures are almost always about state and ordering: which node
held a lock, when it
+ * was renewed, who decided it had expired, what the store actually contained.
The scheduler already
+ * logs all of that as {@code LOCK-DIAG} lines, but only at DEBUG, and CI does
not run at DEBUG --
+ * so every intermittent failure historically arrived as a bare assertion
message with the evidence
+ * discarded. Re-running with {@code -DTEST_LOG_LEVEL=DEBUG} rarely helps,
because an intermittent
+ * failure usually does not recur on demand.
+ * <p>
+ * This extension therefore captures DEBUG for the scheduler packages into a
bounded in-memory ring
+ * buffer that costs nothing on a passing test, and dumps it -- together with
a snapshot of every
+ * task document in the store -- at the moment a test fails. The snapshot is
taken from
+ * {@link TestExecutionExceptionHandler}, which runs before the test's own
{@code @AfterEach}
+ * teardown, so the store is still alive and holds the state that caused the
failure rather than
+ * whatever cleanup left behind.
+ * <p>
+ * When {@code -DTEST_LOG_LEVEL} is set explicitly, the extension stays out of
the way and leaves
+ * logback's configured behaviour alone: an explicit request for console
output should get console
+ * output.
+ */
+public class SchedulerDiagnosticsExtension
+ implements BeforeEachCallback, AfterEachCallback,
TestExecutionExceptionHandler {
+
+ /** Enough lines to cover several checker ticks across a handful of nodes.
*/
+ private static final int BUFFER_SIZE = 4000;
+
+ /** Packages whose DEBUG output explains scheduler behaviour. */
+ private static final String[] CAPTURED_LOGGERS = {
+ "org.apache.unomi.services.impl.scheduler",
+ "org.apache.unomi.services.impl.cluster"
+ };
+
+ private static final String APPENDER_NAME =
"scheduler-diagnostics-ring-buffer";
+ private static final ExtensionContext.Namespace NAMESPACE =
+ ExtensionContext.Namespace.create(SchedulerDiagnosticsExtension.class);
+
+ private static boolean explicitLogLevelRequested() {
+ String requested = System.getProperty("TEST_LOG_LEVEL");
+ return requested != null && !requested.trim().isEmpty();
+ }
+
+ @Override
+ public void beforeEach(ExtensionContext context) {
+ if (explicitLogLevelRequested()) {
+ return;
+ }
+ if (!(LoggerFactory.getILoggerFactory() instanceof LoggerContext)) {
+ return; // not logback (shaded/OSGi runs); nothing to attach to
+ }
+ LoggerContext loggerContext = (LoggerContext)
LoggerFactory.getILoggerFactory();
+
+ CyclicBufferAppender<ILoggingEvent> buffer = new
CyclicBufferAppender<>();
+ buffer.setContext(loggerContext);
+ buffer.setName(APPENDER_NAME);
+ buffer.setMaxSize(BUFFER_SIZE);
+ buffer.start();
+
+ for (String name : CAPTURED_LOGGERS) {
+ ch.qos.logback.classic.Logger logger =
loggerContext.getLogger(name);
+ // additive=false keeps the captured DEBUG out of the console on
passing runs; the
+ // buffer dump below is the only consumer, and it only fires on
failure.
+ logger.setLevel(Level.DEBUG);
+ logger.setAdditive(false);
+ logger.addAppender(buffer);
+ }
+ context.getStore(NAMESPACE).put(APPENDER_NAME, buffer);
+ }
+
+ @Override
+ public void afterEach(ExtensionContext context) {
+ @SuppressWarnings("unchecked")
+ CyclicBufferAppender<ILoggingEvent> buffer =
+ context.getStore(NAMESPACE).remove(APPENDER_NAME,
CyclicBufferAppender.class);
+ if (buffer == null || !(LoggerFactory.getILoggerFactory() instanceof
LoggerContext)) {
+ return;
+ }
+ LoggerContext loggerContext = (LoggerContext)
LoggerFactory.getILoggerFactory();
+ for (String name : CAPTURED_LOGGERS) {
+ ch.qos.logback.classic.Logger logger =
loggerContext.getLogger(name);
+ logger.detachAppender(buffer);
+ logger.setAdditive(true);
+ logger.setLevel(null); // inherit from root again
+ }
+ buffer.stop();
+ }
+
+ @Override
+ public void handleTestExecutionException(ExtensionContext context,
Throwable throwable)
+ throws Throwable {
+ StringBuilder report = new StringBuilder();
+ report.append("\n================ SCHEDULER DIAGNOSTICS for ")
+ .append(context.getRequiredTestClass().getSimpleName()).append('.')
+ .append(context.getRequiredTestMethod().getName())
+ .append(" ================\n")
+ .append("Failure: ").append(throwable).append('\n');
+
+ appendTaskSnapshot(report, context);
+ appendBufferedLog(report, context);
+
+ report.append("================ END SCHEDULER DIAGNOSTICS
================\n");
+ // stdout, not a logger: this must survive whatever logging
configuration is in force, and
+ // Surefire captures stdout into the report the CI log shows.
+ System.out.println(report);
+
+ throw throwable;
+ }
+
+ /**
+ * Dumps every task document the test's persistence service can see. Taken
before teardown, so
+ * this is the state that produced the failure.
+ */
+ private void appendTaskSnapshot(StringBuilder report, ExtensionContext
context) {
+ report.append("\n-- task documents in the store at failure time --\n");
+ PersistenceService persistenceService =
findPersistenceService(context);
+ if (persistenceService == null) {
+ report.append(" (no PersistenceService field found on the test
instance)\n");
+ return;
+ }
+ try {
+ // getAllItems is search-based, and both the in-memory harness and
a real cluster hold a
+ // refresh interval behind the store. Force visibility first: the
test has already
+ // failed, so there is no state left worth preserving, and a
snapshot that silently
+ // reports "(none)" because of refresh lag is worse than useless.
+ try {
+ persistenceService.refreshIndex(ScheduledTask.class);
+ persistenceService.refresh();
+ } catch (Exception ignored) {
+ report.append(" (refresh before snapshot failed; list may lag
the store)\n");
+ }
+ List<ScheduledTask> tasks =
+ persistenceService.getAllItems(ScheduledTask.class, 0, -1,
null).getList();
+ if (tasks.isEmpty()) {
+ report.append(" (none)\n");
+ return;
+ }
+ SimpleDateFormat fmt = new SimpleDateFormat("HH:mm:ss.SSS");
+ for (ScheduledTask task : tasks) {
+ report.append(" ").append(task.getItemId())
+ .append(" type=").append(task.getTaskType())
+ .append(" status=").append(task.getStatus())
+ .append(" enabled=").append(task.isEnabled())
+ .append(" execNode=").append(task.getExecutingNodeId())
+ .append(" lockOwner=").append(task.getLockOwner())
+ .append(" lockDate=")
+ .append(task.getLockDate() == null ? "null" :
fmt.format(task.getLockDate()))
+ .append("
lease=").append(task.getLockLeaseMillis()).append("ms")
+ .append(" success=").append(task.getSuccessCount())
+ .append(" failure=").append(task.getFailureCount())
+ .append(" nextExec=")
+ .append(task.getNextScheduledExecution() == null
+ ? "null" :
fmt.format(task.getNextScheduledExecution()))
+ .append(" history=").append(historySize(task))
+ .append(" lastError=").append(task.getLastError())
+ .append('\n');
+ }
+ } catch (Exception e) {
+ report.append(" (failed to read tasks: ").append(e).append(")\n");
+ }
+ }
+
+ private static int historySize(ScheduledTask task) {
+ Map<String, Object> details = task.getStatusDetails();
+ if (details == null) {
+ return 0;
+ }
+ Object history = details.get("executionHistory");
+ return history instanceof Collection ? ((Collection<?>)
history).size() : 0;
+ }
+
+ /**
+ * Finds a {@link PersistenceService} on the test instance. Reflection
rather than an interface
+ * the tests must implement: the point is that adding this extension to a
test class costs one
+ * annotation and no other change.
+ */
+ private PersistenceService findPersistenceService(ExtensionContext
context) {
+ Object testInstance = context.getTestInstance().orElse(null);
+ if (testInstance == null) {
+ return null;
+ }
+ for (Class<?> type = testInstance.getClass(); type != null; type =
type.getSuperclass()) {
+ for (Field field : type.getDeclaredFields()) {
+ if
(!PersistenceService.class.isAssignableFrom(field.getType())) {
+ continue;
+ }
+ try {
+ field.setAccessible(true);
+ PersistenceService value = (PersistenceService)
field.get(testInstance);
+ if (value != null) {
+ return value;
+ }
+ } catch (ReflectiveOperationException | RuntimeException
ignored) {
+ // Not readable; keep looking.
+ }
+ }
+ }
+ return null;
+ }
+
+ private void appendBufferedLog(StringBuilder report, ExtensionContext
context) {
+ @SuppressWarnings("unchecked")
+ CyclicBufferAppender<ILoggingEvent> buffer =
+ context.getStore(NAMESPACE).get(APPENDER_NAME,
CyclicBufferAppender.class);
+ if (buffer == null) {
+ report.append("\n-- captured scheduler DEBUG log --\n")
+ .append(" (not captured; -DTEST_LOG_LEVEL was set, so the log
went to the console)\n");
+ return;
+ }
+ int count = buffer.getLength();
+ report.append("\n-- captured scheduler DEBUG log (last ").append(count)
+ .append(" events, newest last) --\n");
+ if (count == 0) {
+ report.append(" (empty)\n");
+ return;
+ }
+ SimpleDateFormat fmt = new SimpleDateFormat("HH:mm:ss.SSS");
+ for (int i = 0; i < count; i++) {
+ ILoggingEvent event = buffer.get(i);
+ if (event == null) {
+ continue;
+ }
+ report.append(" ").append(fmt.format(new
Date(event.getTimeStamp())))
+ .append(" [").append(event.getThreadName()).append("] ")
+ .append(event.getLevel()).append(' ')
+ .append(shortLoggerName(event.getLoggerName())).append(" - ")
+ .append(event.getFormattedMessage()).append('\n');
+ }
+ }
+
+ private static String shortLoggerName(String loggerName) {
+ int lastDot = loggerName.lastIndexOf('.');
+ return lastDot < 0 ? loggerName : loggerName.substring(lastDot + 1);
+ }
+}
diff --git
a/services/src/test/java/org/apache/unomi/services/impl/scheduler/SchedulerServiceClusterRaceTest.java
b/services/src/test/java/org/apache/unomi/services/impl/scheduler/SchedulerServiceClusterRaceTest.java
index 62ae062bd..d2ce88f7f 100644
---
a/services/src/test/java/org/apache/unomi/services/impl/scheduler/SchedulerServiceClusterRaceTest.java
+++
b/services/src/test/java/org/apache/unomi/services/impl/scheduler/SchedulerServiceClusterRaceTest.java
@@ -75,6 +75,7 @@ import static org.mockito.Mockito.when;
@ExtendWith(MockitoExtension.class)
@MockitoSettings(strictness = Strictness.LENIENT)
@Tag("ClusterTests")
+@ExtendWith(SchedulerDiagnosticsExtension.class)
public class SchedulerServiceClusterRaceTest {
private static final Logger LOGGER =
LoggerFactory.getLogger(SchedulerServiceClusterRaceTest.class);
@@ -331,6 +332,150 @@ public class SchedulerServiceClusterRaceTest {
assertEquals(1, executors.size(), "Exactly one node should have
executed");
}
+ /**
+ * A node configured with a SHORTER lock timeout than a peer must not
"recover" that peer's
+ * live, renewed lock.
+ * <p>
+ * The owner renews its lock every {@code lockTimeout/3} — a cadence
derived from its OWN
+ * timeout. Before lock leases were recorded ({@link
ScheduledTask#getLockLeaseMillis()}),
+ * expiry was judged against the <em>observer's</em> timeout, so an
observer whose timeout was
+ * shorter than the owner's renewal cadence saw every renewal gap as an
expired lock: it marked
+ * the live execution CRASHED and cleared the lock, and the next peer tick
re-dispatched the
+ * task while the original execution was still running. This reproduced
deterministically as
+ * {@code maxConcurrent=2} with a 1s-timeout observer against 10s-timeout
workers, and is also a
+ * production hazard under config drift or rolling upgrades. The recorded
lease makes expiry
+ * owner-relative, so the divergent observer becomes harmless.
+ */
+ @Test
+ public void testShortTimeoutObserverCannotRecoverLiveRenewedLock() throws
Exception {
+ SchedulerServiceImpl worker1 = createNode("lease-worker1", true,
10000);
+ SchedulerServiceImpl worker2 = createNode("lease-worker2", true,
10000);
+ // Divergent config: this node judges everything with a 500ms timeout.
It registers no
+ // executor for the task type, so any double execution must come via a
worker re-dispatch.
+ SchedulerServiceImpl watchdog = createNode("lease-watchdog", true,
500);
+ seedActiveNodes("lease-worker1", "lease-worker2", "lease-watchdog");
+
+ CountDownLatch started = new CountDownLatch(1);
+ CountDownLatch release = new CountDownLatch(1);
+ AtomicInteger executions = new AtomicInteger(0);
+
+ TaskExecutor executor = new TaskExecutor() {
+ @Override
+ public String getTaskType() {
+ return "lease-liveness-test";
+ }
+
+ @Override
+ public void execute(ScheduledTask task, TaskStatusCallback
callback) throws Exception {
+ executions.incrementAndGet();
+ started.countDown();
+ assertTrue(release.await(TEST_TIMEOUT_MS,
TimeUnit.MILLISECONDS));
+ callback.complete();
+ }
+ };
+ worker1.registerTaskExecutor(executor);
+ worker2.registerTaskExecutor(executor);
+
+ ScheduledTask task = worker1.newTask("lease-liveness-test")
+ .disallowParallelExecution()
+ .asOneShot()
+ .schedule();
+
+ assertTrue(started.await(TEST_TIMEOUT_MS, TimeUnit.MILLISECONDS), "One
worker should start the task");
+
+ // Let the lock age past the watchdog's 500ms timeout while staying
far inside the owner's
+ // 10s lease (the owner's renewal cadence is 10s/3, so the age check
below cannot be
+ // satisfied by a renewal racing us — any observed age > 600ms is a
genuine renewal gap).
+ long deadline = System.currentTimeMillis() + TEST_TIMEOUT_MS;
+ while (System.currentTimeMillis() < deadline) {
+ ScheduledTask stored = persistenceService.load(task.getItemId(),
ScheduledTask.class);
+ if (stored != null && stored.getLockDate() != null
+ && System.currentTimeMillis() -
stored.getLockDate().getTime() > 600) {
+ break;
+ }
+ Thread.sleep(50);
+ }
+
+ // Force the divergent observer's recovery pass repeatedly — the
deterministic version of
+ // the background tick that used to steal the lock.
+ for (int i = 0; i < 3; i++) {
+ watchdog.recoverCrashedTasks();
+ }
+
+ ScheduledTask observed = persistenceService.load(task.getItemId(),
ScheduledTask.class);
+ assertEquals(ScheduledTask.TaskStatus.RUNNING, observed.getStatus(),
+ "A live, renewed lock must not be marked CRASHED by a
shorter-timeout observer");
+ assertNotNull(observed.getLockOwner(), "The owner's lock must not be
cleared");
+
+ release.countDown();
+
+ ScheduledTask done = waitForStatus(worker1, task.getItemId(),
ScheduledTask.TaskStatus.COMPLETED, TEST_TIMEOUT_MS);
+ assertEquals(ScheduledTask.TaskStatus.COMPLETED, done.getStatus());
+ assertEquals(1, executions.get(),
+ "The task must execute exactly once despite the divergent-timeout
observer");
+ }
+
+ /**
+ * The recovery-enabling direction of lease-based expiry: a genuinely DEAD
owner must still be
+ * recovered, and the moment that happens is decided by the lease the dead
owner recorded, not
+ * by the survivor's own (here much longer) timeout. This is the guarantee
that keeps crash
+ * failover working after the lease change — and it is now faster when the
dead node ran with
+ * a short timeout, because peers no longer wait out their own longer
opinion.
+ */
+ @Test
+ public void
testDeadOwnersShortLeaseDrivesPromptRecoveryByPatientSurvivor() throws
Exception {
+ SchedulerServiceImpl survivor = createNode("lease-survivor", true,
30_000);
+ seedActiveNodes("lease-survivor");
+
+ CountDownLatch recovered = new CountDownLatch(1);
+ TaskExecutor executor = new TaskExecutor() {
+ @Override
+ public String getTaskType() {
+ return "dead-owner-lease-test";
+ }
+
+ @Override
+ public void execute(ScheduledTask task, TaskStatusCallback
callback) {
+ recovered.countDown();
+ callback.complete();
+ }
+ };
+ survivor.registerTaskExecutor(executor);
+
+ // Manufacture what a crashed node leaves behind: RUNNING, locked,
lease recorded from a
+ // short timeout, and silent (no renewal will ever come). lockDate is
backdated past the
+ // lease so the very first recovery pass can act.
+ ScheduledTask ghost = new ScheduledTask();
+ ghost.setItemId("ghost-owned-task");
+ ghost.setTaskType("dead-owner-lease-test");
+ ghost.setEnabled(true);
+ ghost.setPersistent(true);
+ ghost.setOneShot(true);
+ ghost.setStatus(ScheduledTask.TaskStatus.RUNNING);
+ ghost.setExecutingNodeId("ghost-node");
+ ghost.setLockOwner("ghost-node");
+ ghost.setLockDate(new Date(System.currentTimeMillis() - 2000));
+ ghost.setLockLeaseMillis(500);
+ persistenceService.save(ghost);
+ persistenceService.refreshIndex(ScheduledTask.class);
+ persistenceService.refresh();
+
+ // Force recovery passes rather than waiting for background ticks. The
survivor's own
+ // timeout is 30s: pre-lease it would have refused to touch this lock
for 30s, and this
+ // latch (10s) would time out. The recorded 500ms lease is what lets
it act now.
+ long deadline = System.currentTimeMillis() + TEST_TIMEOUT_MS;
+ while (recovered.getCount() > 0 && System.currentTimeMillis() <
deadline) {
+ survivor.recoverCrashedTasks();
+ recovered.await(250, TimeUnit.MILLISECONDS);
+ }
+
+ assertTrue(recovered.getCount() == 0,
+ "a patient survivor must recover a dead owner's task as soon as
the OWNER's lease expires");
+ ScheduledTask done = waitForStatus(survivor, "ghost-owned-task",
ScheduledTask.TaskStatus.COMPLETED, TEST_TIMEOUT_MS);
+ assertEquals(ScheduledTask.TaskStatus.COMPLETED, done.getStatus(),
+ "the recovered task must run to completion on the survivor");
+ }
+
@Test
public void testAffinityOpenFieldAfterBackupWindowsWhenPrimaryDead()
throws Exception {
SchedulerServiceImpl backup1 = createNode("aff-backup1", true, 10000);
diff --git
a/services/src/test/java/org/apache/unomi/services/impl/scheduler/SchedulerServiceImplTest.java
b/services/src/test/java/org/apache/unomi/services/impl/scheduler/SchedulerServiceImplTest.java
index d3e91d56c..0521f476b 100644
---
a/services/src/test/java/org/apache/unomi/services/impl/scheduler/SchedulerServiceImplTest.java
+++
b/services/src/test/java/org/apache/unomi/services/impl/scheduler/SchedulerServiceImplTest.java
@@ -74,9 +74,30 @@ import static org.mockito.Mockito.when;
* - RetryTests: Task retry behavior and delay
* - MaintenanceTests: Task cleanup and maintenance
* - QueryTests: Task querying and filtering
+ *
+ * <h3>Debugging</h3>
+ * Logging is configured by {@code src/test/resources/logback-test.xml}; run
with
+ * {@code -DTEST_LOG_LEVEL=DEBUG} to see the scheduler's {@code LOCK-DIAG}
traces, which record
+ * every lock acquisition, renewal, expiry verdict and recovery decision. A
bare assertion
+ * failure from this suite is rarely diagnosable without them.
+ *
+ * <h3>Timing rules for this suite</h3>
+ * The {@code setUp()} scheduler keeps polling in the background for the whole
test, so:
+ * <ul>
+ * <li>Multi-node tests that do not use the setUp scheduler must {@code
preDestroy()} it first —
+ * otherwise it participates in the shared persistence store as an
extra, unaccounted node
+ * (see {@code testNodeFailure} for the pattern).</li>
+ * <li>Never assert an exact execution count while the task can still fire:
cancel the task or
+ * stop the scheduler first, or assert a lower bound.</li>
+ * <li>Prefer latches the test releases over {@code Thread.sleep(N)} for
"keep the executor busy
+ * while I check something" — a fixed sleep is a bet on scheduler timing
that loaded CI
+ * runners lose. Sleeps are acceptable as poll intervals inside bounded
retry loops and as
+ * genuine workload where the duration itself is the test subject.</li>
+ * </ul>
*/
@ExtendWith(MockitoExtension.class)
@MockitoSettings(strictness = Strictness.LENIENT)
+@ExtendWith(SchedulerDiagnosticsExtension.class)
public class SchedulerServiceImplTest {
private static final Logger LOGGER =
LoggerFactory.getLogger(SchedulerServiceImplTest.class);
@@ -89,7 +110,17 @@ public class SchedulerServiceImplTest {
private static final long TEST_TIMEOUT = 15000; // 15 seconds — extra
margin for loaded CI runners
/** Time unit for test timeouts */
private static final TimeUnit TEST_TIME_UNIT = TimeUnit.MILLISECONDS;
- /** Lock timeout for testing lock expiration */
+ /**
+ * Lock timeout for the setUp scheduler and for multi-node tests, matching
+ * {@code TaskLockManager}'s production default. Deliberately NOT short:
the setUp scheduler
+ * keeps polling in the background during every test, and a node whose
lock timeout is shorter
+ * than a peer's renewal cadence (peer timeout / 3) declares that peer's
live locks expired in
+ * the gap between renewals — before lock leases this stole locks from
mid-execution tasks and
+ * double-ran them (the CI flake in testConcurrentLockAcquisition). All
nodes sharing one store
+ * must agree on this value unless lock expiry itself is the behaviour
under test.
+ */
+ private static final long DEFAULT_LOCK_TIMEOUT = 10000; // 10 seconds
+ /** Short lock timeout for tests that exercise lock expiration; set it
explicitly per test. */
private static final long TEST_LOCK_TIMEOUT = 1000; // 1 second
/** Thread pool size for parallel execution */
private static final int TEST_THREAD_POOL_SIZE = 4;
@@ -113,18 +144,12 @@ public class SchedulerServiceImplTest {
// Test categories with documentation
// JUnit 5 provides tags; marker interfaces removed
-
- private static void configureDebugLogging() {
- // Enable debug logging for scheduler package
-
System.setProperty("org.slf4j.simpleLogger.log.org.apache.unomi.services.impl.scheduler",
"DEBUG");
- System.setProperty("org.slf4j.simpleLogger.showDateTime", "true");
- System.setProperty("org.slf4j.simpleLogger.dateTimeFormat",
"yyyy-MM-dd HH:mm:ss.SSS");
- System.setProperty("org.slf4j.simpleLogger.showThreadName", "true");
- }
+ // (An earlier configureDebugLogging() helper set org.slf4j.simpleLogger.*
properties here;
+ // it was dead code — logback-test.xml binds logback, which ignores those.
Use
+ // -DTEST_LOG_LEVEL=DEBUG instead, see the class javadoc.)
@BeforeEach
public void setUp() throws IOException {
- configureDebugLogging();
CustomObjectMapper.getCustomInstance().registerBuiltInItemTypeClass(ScheduledTask.ITEM_TYPE,
ScheduledTask.class);
securityService = TestHelper.createSecurityService();
@@ -155,9 +180,10 @@ public class SchedulerServiceImplTest {
false,
0); // Set TTL to 0 for immediate purging in tests
- // Configure scheduler for testing
+ // Configure scheduler for testing. The lock timeout matches the
production default and
+ // the multi-node tests' nodes; tests exercising expiry shorten it
themselves.
schedulerService.setThreadPoolSize(TEST_THREAD_POOL_SIZE);
- schedulerService.setLockTimeout(TEST_LOCK_TIMEOUT);
+ schedulerService.setLockTimeout(DEFAULT_LOCK_TIMEOUT);
schedulerService.postConstruct();
}
@@ -284,7 +310,10 @@ public class SchedulerServiceImplTest {
.schedule();
assertTrue(executionLatch.await(TEST_TIMEOUT, TEST_TIME_UNIT), "Task
should execute three times");
- assertEquals(3, executionCount.get(), "Task should execute exactly
three times");
+ // Lower bound, not equality: the periodic task keeps firing between
the latch release
+ // and this line, so an exact count is a race against the next period
(cf. the fixed-rate
+ // test above, which already asserts >= for the same reason).
+ assertTrue(executionCount.get() >= 3, "Task should execute at least
three times");
if (workerError.get() != null) {
throw new AssertionError("Assertion failed in worker thread",
workerError.get());
}
@@ -452,7 +481,10 @@ public class SchedulerServiceImplTest {
@Test
@Tag("ClusterTests")
public void testClusteringSupport() throws Exception {
- // Test clustering behavior with multiple nodes
+ // Test clustering behavior with multiple nodes. The setUp scheduler
is not part of this
+ // cluster: stop it so it cannot interfere with the three nodes' tasks
or the node
+ // detection markers below (testNodeFailure pattern).
+ schedulerService.preDestroy();
SchedulerServiceImpl node1 =
TestHelper.createSchedulerService("node1", persistenceService,
executionContextManager, bundleContext, clusterService, -1, true, true);
SchedulerServiceImpl node2 =
TestHelper.createSchedulerService("node2", persistenceService,
executionContextManager, bundleContext, clusterService, -1, true, true);
SchedulerServiceImpl nonExecutorNode =
TestHelper.createSchedulerService("node3", persistenceService,
executionContextManager, bundleContext, clusterService, -1, false, true);
@@ -470,7 +502,10 @@ public class SchedulerServiceImplTest {
persistenceService.refresh();
CountDownLatch exclusiveLatch = new CountDownLatch(1);
- CountDownLatch allNodesLatch = new CountDownLatch(3); // one
execution observed per node
+ // Opens on the FIRST runOnAllNodes execution, on whichever node
wins the first
+ // round; allNodesNodes records the distinct winners (see the
comment below on why
+ // "all three nodes" is not a property the implementation
promises).
+ CountDownLatch allNodesLatch = new CountDownLatch(1);
Set<String> exclusiveNodes = ConcurrentHashMap.newKeySet();
Set<String> allNodesNodes = ConcurrentHashMap.newKeySet();
@@ -496,9 +531,8 @@ public class SchedulerServiceImplTest {
@Override
public void execute(ScheduledTask task, TaskStatusCallback
callback) {
- if (allNodesNodes.add(task.getExecutingNodeId())) {
- allNodesLatch.countDown();
- }
+ allNodesNodes.add(task.getExecutingNodeId());
+ allNodesLatch.countDown();
callback.complete();
}
};
@@ -531,13 +565,22 @@ public class SchedulerServiceImplTest {
exclusiveNodes.contains("node3"),
"Exclusive task must not execute on a non-executor node");
+ // What runOnAllNodes actually promises, as implemented: ANY node
- including a
+ // non-executor - may poll and run the task. It does NOT promise
that every node runs
+ // it: all nodes share the task's single schedule (one
lastExecutionDate /
+ // nextScheduledExecution on one document), so each period has ONE
phase-dependent
+ // winner and there is no fairness across nodes. This test used to
demand an execution
+ // from all three nodes within the timeout, which made it a
lottery over checker-tick
+ // phases - the "runOnAllNodes task should execute on every node"
CI flake. The
+ // non-executor half of the guarantee is pinned deterministically
in
+ // testRunOnAllNodesExecutesOnNonExecutorNode, where the
non-executor is the only node.
assertTrue(
allNodesLatch.await(TEST_TIMEOUT, TEST_TIME_UNIT),
- "runOnAllNodes task should execute on every node including
non-executors");
- assertTrue(allNodesNodes.contains("node1"), "runOnAllNodes should
run on node1");
- assertTrue(allNodesNodes.contains("node2"), "runOnAllNodes should
run on node2");
- assertTrue(allNodesNodes.contains("node3"), "runOnAllNodes should
run on non-executor node3");
+ "runOnAllNodes task should execute on at least one node");
+ // Keep the lock-inspection task's execution alive until this test
has finished
+ // inspecting its lock, instead of betting on a fixed sleep
outlasting the checks.
+ CountDownLatch lockTaskRelease = new CountDownLatch(1);
TaskExecutor clusterLockTestExecutor = new TaskExecutor() {
@Override
public String getTaskType() {
@@ -546,7 +589,7 @@ public class SchedulerServiceImplTest {
@Override
public void execute(ScheduledTask task, TaskStatusCallback
callback) {
try {
- Thread.sleep(5000);
+ lockTaskRelease.await(TEST_TIMEOUT, TEST_TIME_UNIT);
callback.complete();
} catch (InterruptedException e) {
callback.fail(e.getMessage());
@@ -554,39 +597,103 @@ public class SchedulerServiceImplTest {
}
};
- schedulerService.registerTaskExecutor(clusterLockTestExecutor);
+ // Register on the cluster's own executor nodes (the setUp
scheduler is stopped).
+ node1.registerTaskExecutor(clusterLockTestExecutor);
+ node2.registerTaskExecutor(clusterLockTestExecutor);
- // Test lock management
- ScheduledTask lockTask = node1.newTask("cluster-lock-test")
- .disallowParallelExecution()
- .schedule();
+ try {
+ // Test lock management
+ ScheduledTask lockTask = node1.newTask("cluster-lock-test")
+ .disallowParallelExecution()
+ .schedule();
+
+ // Refresh persistence to ensure task updates are
available (handles refresh delay)
+ persistenceService.refresh();
+ // Wait until the task has a lock owner. Deadline-based rather
than
+ // TestHelper.retryUntil's fixed 20x100ms budget, which a
loaded runner exceeds
+ // (dispatch needs a checker tick plus the simulated refresh
delay).
+ ScheduledTask lockedTask = null;
+ long lockDeadline = System.currentTimeMillis() + TEST_TIMEOUT;
+ while (System.currentTimeMillis() < lockDeadline) {
+ lockedTask = persistenceService.load(lockTask.getItemId(),
ScheduledTask.class);
+ if (lockedTask != null && lockedTask.getLockOwner() !=
null) {
+ break;
+ }
+ Thread.sleep(100);
+ }
+ assertNotNull(lockedTask, "Lock task should be persisted");
+ assertNotNull(lockedTask.getLockOwner(), "Task should have
lock owner");
+ assertNotNull(lockedTask.getLockDate(), "Task should have lock
date");
+
+ // Test lock release - directly update task in persistence
+ lockedTask.setLockOwner(null);
+ lockedTask.setLockDate(null);
+ lockedTask.setLockLeaseMillis(0);
+ persistenceService.save(lockedTask);
+
+ // Refresh index to ensure changes are visible
+ persistenceService.refreshIndex(ScheduledTask.class);
+
+ // Get latest state and verify lock release
+ ScheduledTask releasedTask =
persistenceService.load(lockTask.getItemId(), ScheduledTask.class);
+ assertNull(releasedTask.getLockOwner(), "Lock should be
released");
+ } finally {
+ lockTaskRelease.countDown();
+ }
- // Refresh persistence to ensure task updates are available
(handles refresh delay)
- persistenceService.refresh();
- // Retry until task has lock owner (handles refresh delay for
updates)
- ScheduledTask lockedTask = TestHelper.retryUntil(
- () -> persistenceService.load(lockTask.getItemId(),
ScheduledTask.class),
- t -> t != null && t.getLockOwner() != null
- );
- assertNotNull(lockedTask.getLockOwner(), "Task should have lock
owner");
- assertNotNull(lockedTask.getLockDate(), "Task should have lock
date");
+ } finally {
+ node1.preDestroy();
+ node2.preDestroy();
+ nonExecutorNode.preDestroy();
+ }
+ }
- // Test lock release - directly update task in persistence
- lockedTask.setLockOwner(null);
- lockedTask.setLockDate(null);
- persistenceService.save(lockedTask);
+ /**
+ * The non-executor half of the runOnAllNodes guarantee, pinned
deterministically: a node
+ * with {@code executorNode=false} must still poll for and execute
runOnAllNodes tasks.
+ * <p>
+ * testClusteringSupport cannot assert this reliably — with executor nodes
present, all nodes
+ * race on the task's single shared schedule and there is no fairness, so
whether the
+ * non-executor ever wins a round within the timeout is checker-phase
luck. Here the
+ * non-executor is the ONLY node, so if it does not poll runOnAllNodes
work (the regression
+ * this pins), nothing executes and the latch times out.
+ */
+ @Test
+ @Tag("ClusterTests")
+ public void testRunOnAllNodesExecutesOnNonExecutorNode() throws Exception {
+ schedulerService.preDestroy();
+ SchedulerServiceImpl nonExecutorOnly =
TestHelper.createSchedulerService(
+ "solo-non-executor", persistenceService, executionContextManager,
bundleContext, clusterService, -1, false, true);
- // Refresh index to ensure changes are visible
- persistenceService.refreshIndex(ScheduledTask.class);
+ try {
+ CountDownLatch executed = new CountDownLatch(1);
+ AtomicReference<String> executingNode = new AtomicReference<>();
- // Get latest state and verify lock release
- ScheduledTask releasedTask =
persistenceService.load(lockTask.getItemId(), ScheduledTask.class);
- assertNull(releasedTask.getLockOwner(), "Lock should be released");
+ TaskExecutor executor = new TaskExecutor() {
+ @Override
+ public String getTaskType() {
+ return "all-nodes-solo-test";
+ }
+ @Override
+ public void execute(ScheduledTask task, TaskStatusCallback
callback) {
+ executingNode.set(task.getExecutingNodeId());
+ executed.countDown();
+ callback.complete();
+ }
+ };
+ nonExecutorOnly.registerTaskExecutor(executor);
+
+ nonExecutorOnly.newTask("all-nodes-solo-test")
+ .runOnAllNodes()
+ .withPeriod(100, TimeUnit.MILLISECONDS)
+ .schedule();
+
+ assertTrue(executed.await(TEST_TIMEOUT, TEST_TIME_UNIT),
+ "a non-executor node must poll for and run runOnAllNodes
tasks");
+ assertEquals("solo-non-executor", executingNode.get());
} finally {
- node1.preDestroy();
- node2.preDestroy();
- nonExecutorNode.preDestroy();
+ nonExecutorOnly.preDestroy();
}
}
@@ -748,17 +855,29 @@ public class SchedulerServiceImplTest {
failureLatch.await(TEST_TIMEOUT, TEST_TIME_UNIT),
"Task should fail once");
- // Verify metrics and history
+ // The 100ms-period task keeps executing (and failing) after the
latches fire, so exact
+ // counts are a race against the next period. Cancel it and wait for
the cancellation to
+ // land before reading anything.
+ schedulerService.cancelTask(task.getItemId());
+ TestHelper.retryUntil(
+ () -> schedulerService.getTask(task.getItemId()),
+ t -> t != null && t.getStatus() != ScheduledTask.TaskStatus.RUNNING
+ && t.getStatus() != ScheduledTask.TaskStatus.SCHEDULED);
+
+ // Verify metrics and history. Successes are exact (the executor only
ever completes the
+ // first two); failures are a lower bound (every later period failed
until the cancel won).
ScheduledTask finalTask = schedulerService.getTask(task.getItemId());
@SuppressWarnings("unchecked")
List<Map<String, Object>> history =
(List<Map<String, Object>>)
finalTask.getStatusDetails().get("executionHistory");
assertNotNull(history, "Should have execution history");
- assertEquals(3, history.size(), "Should have 3 history entries");
+ assertTrue(history.size() >= 3, "Should have at least 3 history
entries, had " + history.size());
assertEquals(2, finalTask.getSuccessCount(), "Should have 2 successful
executions");
- assertEquals(1, finalTask.getFailureCount(), "Should have 1 failed
execution");
- assertEquals(3, finalTask.getSuccessCount() +
finalTask.getFailureCount(), "Total executions should be 3");
+ assertTrue(finalTask.getFailureCount() >= 1,
+ "Should have at least 1 failed execution, had " +
finalTask.getFailureCount());
+ // No history-size == successCount+failureCount equality here: an
execution in flight
+ // while the cancel lands may or may not get its failure recorded, by
design.
// Verify history entries
int successEntries = 0;
@@ -774,7 +893,7 @@ public class SchedulerServiceImplTest {
}
assertEquals(2, successEntries, "Should have 2 successful executions");
- assertEquals(1, failureEntries, "Should have 1 failed execution");
+ assertTrue(failureEntries >= 1, "Should have at least 1 failed
execution");
// Verify metrics
assertTrue(schedulerService.getMetric("tasks.completed") > 0, "Should
have completed tasks metric");
@@ -932,12 +1051,34 @@ public class SchedulerServiceImplTest {
executionLatch.await(TEST_TIMEOUT, TimeUnit.MILLISECONDS),
"Task should complete all executions");
- // Verify retry delays
+ // Verify retry delays. Deliberately NOT relaxed: an execution landing
sooner than the
+ // retry delay means a retry attempt was dispatched early, which is a
real contract
+ // violation worth failing on. Suspected mechanism if this fires on CI
and not locally:
+ // prepareForExecution() checks due-ness against the task instance it
was handed, and the
+ // checker discovers tasks with a search query that lags the store, so
a stale copy still
+ // carrying the pre-retry (already past) nextScheduledExecution passes
the due check and
+ // executes immediately. Unproven - hence the diagnostics below rather
than a weakened
+ // assertion, so the next occurrence is decisive instead of just a
boolean.
for (int i = 1; i < executionTimes.size(); i++) {
- long delay = executionTimes.get(i) - executionTimes.get(i-1);
- assertTrue(
- delay >= TEST_RETRY_DELAY,
- "Retry delay should be at least " + TEST_RETRY_DELAY + "ms");
+ long delay = executionTimes.get(i) - executionTimes.get(i - 1);
+ if (delay < TEST_RETRY_DELAY) {
+ StringBuilder detail = new StringBuilder();
+ detail.append("Retry delay should be at least
").append(TEST_RETRY_DELAY)
+ .append("ms but execution #").append(i + 1).append(" came
").append(delay)
+ .append("ms after #").append(i)
+ .append(". persistent=").append(persistent)
+ .append(", executions=").append(executionTimes.size())
+ .append(" (expected ").append(TEST_MAX_RETRIES +
1).append("), gaps=[");
+ for (int j = 1; j < executionTimes.size(); j++) {
+ if (j > 1) {
+ detail.append(", ");
+ }
+ detail.append(executionTimes.get(j) - executionTimes.get(j
- 1)).append("ms");
+ }
+ detail.append("]. More executions than expected points at a
duplicate dispatch; "
+ + "the right count with a short gap points at an early
retry schedule.");
+ fail(detail.toString());
+ }
}
// Wait for the task to transition from RUNNING to COMPLETED state
@@ -1023,13 +1164,24 @@ public class SchedulerServiceImplTest {
* tasks that already executed, stranding the task in CRASHED state
forever. The
* execution manager must recognize that the execution it owns is still
alive, reclaim
* the task and process the failure (and its retry) normally.
+ *
+ * <p>NOTE: since lock renewal was introduced (the lock is re-stamped
every lockTimeout/3
+ * while the executor runs), a stalled-but-live execution's lock no longer
expires from
+ * natural timing, so the CRASH-mark this test was written around does not
fire anymore -
+ * verified from the LOCK-DIAG traces: renewal succeeds throughout the
stall and no expiry
+ * verdict ever triggers. The test remains valuable as a pin on the
surviving behaviour
+ * (a failure reported after a stall longer than the lock timeout still
schedules its
+ * retries and completes), and its assertions were already written to
tolerate both worlds
+ * (>= 3 executions). The reclaim path itself is now only reachable when
renewal genuinely
+ * stops (e.g. a GC pause longer than the full lease) and is covered at
unit level in
+ * TaskExecutionManagerTest.
*/
@Test
@Tag("RetryTests")
public void testOneShotRetryAfterRecoveryMarksLiveExecutionCrashed()
throws Exception {
- // setUp() only sets the lock timeout on the scheduler service; the
lock manager
- // created by TestHelper keeps its 10s default. Shorten it here so a
stalled
- // execution's lock actually expires within this test's stall window.
+ // Shorten the lock timeout so the stall below dwarfs it.
(setLockTimeout on the service
+ // propagates to the lock manager as well; setting the lock manager
directly is
+ // equivalent and kept for clarity about what the timeout is FOR here.)
schedulerService.getLockManager().setLockTimeout(TEST_LOCK_TIMEOUT);
CountDownLatch completionLatch = new CountDownLatch(1);
@@ -1320,6 +1472,7 @@ public class SchedulerServiceImplTest {
schedulerService.setLockTimeout(TEST_LOCK_TIMEOUT);
CountDownLatch executionLatch = new CountDownLatch(1);
+ CountDownLatch holdRelease = new CountDownLatch(1);
AtomicBoolean taskStarted = new AtomicBoolean(false);
TaskExecutor executor = new TaskExecutor() {
@@ -1335,8 +1488,9 @@ public class SchedulerServiceImplTest {
taskStarted.set(true);
executionLatch.countDown();
- // Hold the lock longer than timeout
- Thread.sleep(TEST_LOCK_TIMEOUT * 2);
+ // Hold the lock until the test has finished inspecting it
- a latch the
+ // test releases, not a fixed sleep the test hopes is long
enough.
+ holdRelease.await(TEST_TIMEOUT, TEST_TIME_UNIT);
callback.complete();
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
@@ -1359,12 +1513,15 @@ public class SchedulerServiceImplTest {
// Directly update task to simulate lock expiration
runningTask.setLockOwner(null);
runningTask.setLockDate(null);
+ runningTask.setLockLeaseMillis(0);
persistenceService.save(runningTask);
persistenceService.refreshIndex(ScheduledTask.class);
// Check lock status after manual release
ScheduledTask updatedTask = persistenceService.load(task.getItemId(),
ScheduledTask.class);
assertNull(updatedTask.getLockOwner(), "Lock should be released after
manual update");
+
+ holdRelease.countDown();
}
/**
@@ -1631,6 +1788,12 @@ public class SchedulerServiceImplTest {
@Test
@Tag("ClusterTests")
public void testConcurrentLockAcquisition() throws Exception {
+ // This test is about node1/node2 only: stop the setUp scheduler so
the "two-node" cluster
+ // really has two nodes (testNodeFailure pattern). It used to stay up
with a 1s lock
+ // timeout against these nodes' 10s, declare their live locks expired
between renewals,
+ // and mark the running task CRASHED — which a peer then re-dispatched
concurrently
+ // (the "expected: <1> but was: <2>" CI flake).
+ schedulerService.preDestroy();
SchedulerServiceImpl node1 =
TestHelper.createSchedulerService("node1", persistenceService,
executionContextManager, bundleContext, clusterService, -1, true, true);
SchedulerServiceImpl node2 =
TestHelper.createSchedulerService("node2", persistenceService,
executionContextManager, bundleContext, clusterService, -1, true, true);
@@ -1738,6 +1901,8 @@ public class SchedulerServiceImplTest {
@Test
@Tag("ClusterTests")
public void testTaskRebalancing() throws Exception {
+ // Two-node test: stop the setUp scheduler so it is not a hidden third
participant.
+ schedulerService.preDestroy();
SchedulerServiceImpl node1 =
TestHelper.createSchedulerService("node1", persistenceService,
executionContextManager, bundleContext, clusterService, -1, true, true);
SchedulerServiceImpl node2 = null;
try {
@@ -1861,6 +2026,8 @@ public class SchedulerServiceImplTest {
@Test
@Tag("ClusterTests")
public void testLockStealing() throws Exception {
+ // Two-node test: stop the setUp scheduler so it is not a hidden third
participant.
+ schedulerService.preDestroy();
SchedulerServiceImpl node1 =
TestHelper.createSchedulerService("node1", persistenceService,
executionContextManager, bundleContext, clusterService, -1, true, true);
SchedulerServiceImpl node2 =
TestHelper.createSchedulerService("node2", persistenceService,
executionContextManager, bundleContext, clusterService, -1, true, true);
@@ -1946,6 +2113,10 @@ public class SchedulerServiceImplTest {
@Test
public void testNodeAffinity() throws Exception {
+ // Three-node test. Stop the setUp scheduler: getActiveNodes() falls
back to scanning
+ // tasks with recent locks, and a foreign recovery pass that clears
the detection tasks'
+ // locks below would silently shrink the cluster this test asserts on.
+ schedulerService.preDestroy();
// Create test nodes with cluster service
SchedulerServiceImpl node1 =
TestHelper.createSchedulerService("node1", persistenceService,
executionContextManager, bundleContext, clusterService, -1, true, true);
SchedulerServiceImpl node2 =
TestHelper.createSchedulerService("node2", persistenceService,
executionContextManager, bundleContext, clusterService, -1, true, true);
@@ -2425,7 +2596,9 @@ public class SchedulerServiceImplTest {
newSchedulerService.preDestroy();
assertTrue(executed, "Task should execute after scheduler restart");
- assertEquals(2, executionCount.get(), "Task should have executed
twice");
+ // Lower bound: the 500ms-period task may legitimately fire again
between the latch
+ // release and preDestroy() completing on a slow runner.
+ assertTrue(executionCount.get() >= 2, "Task should have executed at
least twice");
// Verify the reloaded task has same ID
ScheduledTask reloadedTask =
persistenceService.load(persistentTask.getItemId(), ScheduledTask.class);
@@ -2757,10 +2930,13 @@ public class SchedulerServiceImplTest {
assertTrue(
secondExecutionLatch.await(TEST_TIMEOUT * 2, TEST_TIME_UNIT),
"Task should execute after restart with dedicated executor");
- assertEquals(2, executionCount.get(), "Task should execute twice");
- // Clean up
+ // Stop the scheduler before asserting the count, exactly like the
first assertion above:
+ // the task runs at 100ms fixed rate, so a third tick can fire between
the latch release
+ // and the assert. After preDestroy() the count is stable; >=
tolerates a tick that
+ // squeezed in before shutdown took effect.
newSchedulerService.preDestroy();
+ assertTrue(executionCount.get() >= 2, "Task should execute at least
twice");
}
/**
diff --git
a/services/src/test/java/org/apache/unomi/services/impl/scheduler/TaskExecutionManagerTest.java
b/services/src/test/java/org/apache/unomi/services/impl/scheduler/TaskExecutionManagerTest.java
index bafcc5356..8aa3105b0 100644
---
a/services/src/test/java/org/apache/unomi/services/impl/scheduler/TaskExecutionManagerTest.java
+++
b/services/src/test/java/org/apache/unomi/services/impl/scheduler/TaskExecutionManagerTest.java
@@ -28,8 +28,12 @@ import org.mockito.junit.jupiter.MockitoExtension;
import org.mockito.junit.jupiter.MockitoSettings;
import org.mockito.quality.Strictness;
+import java.util.ArrayList;
import java.util.Collections;
import java.util.Date;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
@@ -45,6 +49,7 @@ import static org.mockito.Mockito.*;
*/
@ExtendWith(MockitoExtension.class)
@MockitoSettings(strictness = Strictness.LENIENT)
+@ExtendWith(SchedulerDiagnosticsExtension.class)
public class TaskExecutionManagerTest {
private static final String NODE = "exec-node";
@@ -91,6 +96,62 @@ public class TaskExecutionManagerTest {
executionManager.shutdown();
}
+ /**
+ * A terminal handler must increment counters from the STORE's values, not
from the possibly
+ * stale copy the wrapper is carrying.
+ * <p>
+ * The dispatch path discovers tasks with a search query, which lags the
store by up to the
+ * index refresh interval, so the executing instance can hold counters
that predate writes
+ * already committed. {@code persistTerminalState}'s compare-and-set
protects only the document
+ * version, so incrementing a stale base then CAS-writing it succeeds and
silently loses the
+ * newer count. Observed in CI as a periodic task reporting one success
after two successful
+ * executions ({@code SchedulerServiceImplTest.testMetricsAndHistory}).
+ */
+ @Test
+ public void testTerminalCompletionRebasesCountersOnStoreValues() throws
Exception {
+ CountDownLatch done = new CountDownLatch(1);
+ TaskExecutor executor = new TaskExecutor() {
+ @Override public String getTaskType() { return "stale-counters"; }
+ @Override public void execute(ScheduledTask task,
TaskStatusCallback callback) {
+ callback.complete();
+ done.countDown();
+ }
+ };
+
+ // What the wrapper carries: a search-lagged view that has not seen
the first success.
+ ScheduledTask stale = TaskTestFixtures.baseTask("stale-counters");
+ stale.setOneShot(false);
+ stale.setPeriod(60_000);
+ stale.setSuccessCount(0);
+ stale.setFailureCount(0);
+
+ // What the store actually holds: one success already recorded, with
its history entry.
+ ScheduledTask store = TaskTestFixtures.baseTask("stale-counters");
+ store.setItemId(stale.getItemId());
+ store.setStatus(ScheduledTask.TaskStatus.RUNNING);
+ store.setExecutingNodeId(NODE);
+ store.setSuccessCount(1);
+ Map<String, Object> storeDetails = new HashMap<>();
+ List<Map<String, Object>> storeHistory = new ArrayList<>();
+ storeHistory.add(Collections.singletonMap("status", "SUCCESS"));
+ storeDetails.put("executionHistory", storeHistory);
+ store.setStatusDetails(storeDetails);
+ when(schedulerService.getTask(eq(stale.getItemId()),
eq(true))).thenReturn(store);
+
+ executionManager.executeTask(stale, executor);
+ assertTrue(done.await(5, TimeUnit.SECONDS));
+ awaitStatus(stale, ScheduledTask.TaskStatus.SCHEDULED, 5000);
+
+ assertEquals(2, stale.getSuccessCount(),
+ "the second success must count from the store's value (1), not the
stale copy's (0)");
+
+ @SuppressWarnings("unchecked")
+ List<Map<String, Object>> history =
+ (List<Map<String, Object>>)
stale.getStatusDetails().get("executionHistory");
+ assertEquals(2, history.size(),
+ "history must extend the store's entries rather than restart from
the stale copy's");
+ }
+
@Test
public void testPrepareForExecutionRejectsDisabledAndWrongStatus() {
ScheduledTask disabled = TaskTestFixtures.baseTask("p");
@@ -128,6 +189,19 @@ public class TaskExecutionManagerTest {
assertEquals(ScheduledTask.TaskStatus.SCHEDULED, task.getStatus());
}
+ /**
+ * Waits for the wrapper's asynchronous terminal transition to land on the
shared task object.
+ * The executor's callback returns before the wrapper finishes its
bookkeeping, so asserting
+ * the final status right after the latch (or after a fixed sleep) races
the wrapper thread.
+ */
+ private static void awaitStatus(ScheduledTask task,
ScheduledTask.TaskStatus expected, long timeoutMs)
+ throws InterruptedException {
+ long deadline = System.currentTimeMillis() + timeoutMs;
+ while (task.getStatus() != expected && System.currentTimeMillis() <
deadline) {
+ Thread.sleep(20);
+ }
+ }
+
@Test
public void testExecuteTaskDuplicateDispatchIsSkipped() throws Exception {
CountDownLatch started = new CountDownLatch(1);
@@ -148,6 +222,9 @@ public class TaskExecutionManagerTest {
// Second dispatch while claim held
executionManager.executeTask(task, executor);
release.countDown();
+ // Deliberate quiet window for a NEGATIVE assertion: a wrongly
accepted duplicate
+ // dispatch would start within milliseconds. Too short can only miss a
violation
+ // (false green), never fail a healthy run.
Thread.sleep(200);
assertEquals(1, runs.get());
}
@@ -273,7 +350,7 @@ public class TaskExecutionManagerTest {
executionManager.executeTask(task, executor);
assertTrue(done.await(5, TimeUnit.SECONDS));
- Thread.sleep(100);
+ awaitStatus(task, ScheduledTask.TaskStatus.FAILED, 5000);
assertEquals(ScheduledTask.TaskStatus.FAILED, task.getStatus());
assertEquals(1, task.getFailureCount());
assertTrue(task.isEnabled());
@@ -298,7 +375,7 @@ public class TaskExecutionManagerTest {
executionManager.executeTask(task, executor);
assertTrue(done.await(5, TimeUnit.SECONDS));
- Thread.sleep(100);
+ awaitStatus(task, ScheduledTask.TaskStatus.SCHEDULED, 5000);
assertEquals(0, task.getFailureCount());
assertEquals(ScheduledTask.TaskStatus.SCHEDULED, task.getStatus());
assertNotNull(task.getNextScheduledExecution());
@@ -340,7 +417,7 @@ public class TaskExecutionManagerTest {
releaser.start();
executionManager.shutdown();
releaser.join(2000);
- Thread.sleep(200);
+ awaitStatus(task, ScheduledTask.TaskStatus.SCHEDULED, 5000);
assertEquals(ScheduledTask.TaskStatus.SCHEDULED, task.getStatus());
assertEquals(1, task.getFailureCount());
// No second attempt — retry schedule skipped after scheduler shutdown
@@ -362,7 +439,7 @@ public class TaskExecutionManagerTest {
executionManager.executeTask(task, executor);
assertTrue(done.await(5, TimeUnit.SECONDS));
- Thread.sleep(100);
+ awaitStatus(task, ScheduledTask.TaskStatus.COMPLETED, 5000);
assertEquals(ScheduledTask.TaskStatus.COMPLETED, task.getStatus());
assertFalse(task.isEnabled());
assertNull(task.getNextScheduledExecution());
@@ -384,7 +461,7 @@ public class TaskExecutionManagerTest {
executionManager.executeTask(task, executor);
assertTrue(done.await(5, TimeUnit.SECONDS));
- Thread.sleep(100);
+ awaitStatus(task, ScheduledTask.TaskStatus.SCHEDULED, 5000);
assertEquals(ScheduledTask.TaskStatus.SCHEDULED, task.getStatus());
assertNotNull(task.getNextScheduledExecution());
assertTrue(task.getNextScheduledExecution().getTime() >= before +
5_000);
@@ -406,7 +483,7 @@ public class TaskExecutionManagerTest {
executionManager.executeTask(task, executor);
assertTrue(done.await(5, TimeUnit.SECONDS));
- Thread.sleep(100);
+ awaitStatus(task, ScheduledTask.TaskStatus.COMPLETED, 5000);
assertEquals(ScheduledTask.TaskStatus.COMPLETED, task.getStatus());
}
@@ -427,6 +504,8 @@ public class TaskExecutionManagerTest {
executionManager.executeTask(task, executor);
assertTrue(done.await(5, TimeUnit.SECONDS));
+ // Deliberate quiet window for a NEGATIVE assertion (callbacks must
have been ignored);
+ // a poll cannot confirm that nothing happened.
Thread.sleep(100);
assertEquals(ScheduledTask.TaskStatus.CANCELLED, task.getStatus());
assertEquals(completedBefore,
metricsManager.getMetric(TaskMetricsManager.METRIC_TASKS_COMPLETED));
@@ -448,7 +527,7 @@ public class TaskExecutionManagerTest {
executionManager.executeTask(task, executor);
assertTrue(done.await(5, TimeUnit.SECONDS));
- Thread.sleep(100);
+ awaitStatus(task, ScheduledTask.TaskStatus.COMPLETED, 5000);
assertEquals(ScheduledTask.TaskStatus.COMPLETED, task.getStatus());
assertFalse(task.isEnabled());
}
@@ -545,12 +624,12 @@ public class TaskExecutionManagerTest {
ScheduledTask task = TaskTestFixtures.baseTask("cancel-race");
executionManager.executeTask(task, executor);
assertTrue(done.await(5, TimeUnit.SECONDS));
- Thread.sleep(100);
- assertEquals(ScheduledTask.TaskStatus.CANCELLED, task.getStatus());
// persistTerminalState() is skipped (terminal transition correctly
bailed out above), but
// the wrapper's cleanup still CAS-clears executingNodeId once; that
write is expected to
// fail harmlessly against a real store since the document moved on to
CANCELLED.
- verify(schedulerService, times(1)).saveTaskWithRefresh(any());
+ // timeout() waits for the asynchronous cleanup instead of betting a
fixed sleep on it.
+ verify(schedulerService,
timeout(5000).times(1)).saveTaskWithRefresh(any());
+ assertEquals(ScheduledTask.TaskStatus.CANCELLED, task.getStatus());
assertEquals(0,
metricsManager.getMetric(TaskMetricsManager.METRIC_TASKS_COMPLETED));
}
@@ -572,12 +651,12 @@ public class TaskExecutionManagerTest {
ScheduledTask task = TaskTestFixtures.baseTask("peer-lock");
executionManager.executeTask(task, executor);
assertTrue(done.await(5, TimeUnit.SECONDS));
- Thread.sleep(100);
- assertEquals(ScheduledTask.TaskStatus.RUNNING, task.getStatus());
// persistTerminalState() is skipped (peer holds the lock), but the
wrapper's cleanup still
// CAS-clears executingNodeId once; that write is expected to fail
harmlessly against a real
// store since the peer is the authoritative owner.
- verify(schedulerService, times(1)).saveTaskWithRefresh(any());
+ // timeout() waits for the asynchronous cleanup instead of betting a
fixed sleep on it.
+ verify(schedulerService,
timeout(5000).times(1)).saveTaskWithRefresh(any());
+ assertEquals(ScheduledTask.TaskStatus.RUNNING, task.getStatus());
}
@Test
@@ -600,9 +679,11 @@ public class TaskExecutionManagerTest {
};
ScheduledTask task = TaskTestFixtures.baseTask("abort-prep");
executionManager.executeTask(task, executor);
- Thread.sleep(300);
- assertEquals(1, executed.getCount(), "executor must not run after
shutdown-abort");
+ // Positive half: wait for the asynchronous abort to land instead of a
fixed sleep.
+ awaitStatus(task, ScheduledTask.TaskStatus.CRASHED, 5000);
assertEquals(ScheduledTask.TaskStatus.CRASHED, task.getStatus());
+ // Negative half: the executor must never have run (green-direction
check).
+ assertEquals(1, executed.getCount(), "executor must not run after
shutdown-abort");
assertNull(task.getLockOwner());
verify(schedulerService,
atLeastOnce()).saveTask(any(ScheduledTask.class), eq(true));
}
diff --git
a/services/src/test/java/org/apache/unomi/services/impl/scheduler/TaskLockManagerTest.java
b/services/src/test/java/org/apache/unomi/services/impl/scheduler/TaskLockManagerTest.java
index 249783884..57cec8d58 100644
---
a/services/src/test/java/org/apache/unomi/services/impl/scheduler/TaskLockManagerTest.java
+++
b/services/src/test/java/org/apache/unomi/services/impl/scheduler/TaskLockManagerTest.java
@@ -44,6 +44,7 @@ import static org.mockito.Mockito.*;
*/
@ExtendWith(MockitoExtension.class)
@MockitoSettings(strictness = Strictness.LENIENT)
+@ExtendWith(SchedulerDiagnosticsExtension.class)
public class TaskLockManagerTest {
private static final String NODE = "lock-node";
@@ -216,6 +217,188 @@ public class TaskLockManagerTest {
assertFalse(lockManager.isLockExpired(task));
}
+ /**
+ * Expiry must be judged against the lease the OWNER recorded with the
lock, not this
+ * observer's own timeout. An observer configured shorter than the owner's
renewal cadence
+ * would otherwise "recover" a live, renewed lock between two renewals and
double-run the
+ * task (see
SchedulerServiceClusterRaceTest#testShortTimeoutObserverCannotRecoverLiveRenewedLock
+ * for the end-to-end version).
+ */
+ @Test
+ public void testIsLockExpiredHonoursRecordedLeaseOverObserverTimeout() {
+ ScheduledTask task = TaskTestFixtures.baseTask("lease");
+ task.setLockDate(new Date(System.currentTimeMillis() - 5000));
+
+ // 5s-old lock, observer timeout 1s: expired by observer maths, but
the owner granted 10s.
+ task.setLockLeaseMillis(10000);
+ assertFalse(lockManager.isLockExpired(task),
+ "a lock inside its owner-recorded lease must not expire under a
shorter observer timeout");
+
+ // The reverse also holds: an owner that granted itself a SHORT lease
is expired even
+ // when the observer's own timeout would still consider it live.
+ task.setLockDate(new Date(System.currentTimeMillis() - 500));
+ task.setLockLeaseMillis(100);
+ assertTrue(lockManager.isLockExpired(task),
+ "a lock past its owner-recorded lease is expired regardless of the
observer timeout");
+ }
+
+ /** Locks written before lease recording (lease 0) fall back to the
observer's own timeout. */
+ @Test
+ public void testIsLockExpiredFallsBackToObserverTimeoutForLegacyLocks() {
+ ScheduledTask task = TaskTestFixtures.baseTask("legacy");
+ task.setLockDate(new Date(System.currentTimeMillis() - 5000));
+ task.setLockLeaseMillis(0);
+ assertTrue(lockManager.isLockExpired(task));
+
+ task.setLockDate(new Date());
+ assertFalse(lockManager.isLockExpired(task));
+ }
+
+ /** A corrupt negative lease must not wedge or widen the lock: treat it
like a legacy lock. */
+ @Test
+ public void testIsLockExpiredNegativeLeaseFallsBackToObserverTimeout() {
+ ScheduledTask task = TaskTestFixtures.baseTask("corrupt");
+ task.setLockDate(new Date(System.currentTimeMillis() - 5000));
+ task.setLockLeaseMillis(-1);
+ assertTrue(lockManager.isLockExpired(task), "negative lease + old
lock: observer timeout applies");
+
+ task.setLockDate(new Date());
+ assertFalse(lockManager.isLockExpired(task), "negative lease + fresh
lock: observer timeout applies");
+ }
+
+ /** Boundary parity with the observer-timeout path: age == lease is NOT
yet expired. */
+ @Test
+ public void testIsLockExpiredFalseWhenAgeEqualsRecordedLease() {
+ ScheduledTask task = TaskTestFixtures.baseTask("edge");
+ task.setLockLeaseMillis(2000);
+ task.setLockDate(new Date(System.currentTimeMillis() - 2000));
+ assertFalse(lockManager.isLockExpired(task));
+ }
+
+ /**
+ * An absurd lease (misconfigured or corrupt owner) must not overflow the
arithmetic. The lock
+ * is honoured as unexpired — the owner declared it, and stealing it risks
double execution;
+ * a genuinely wedged task from a dead misconfigured node is an operator
decision, not one a
+ * peer may take unilaterally with a shorter opinion.
+ */
+ @Test
+ public void testIsLockExpiredHugeLeaseIsHonouredWithoutOverflow() {
+ ScheduledTask task = TaskTestFixtures.baseTask("huge");
+ task.setLockLeaseMillis(Long.MAX_VALUE);
+ task.setLockDate(new Date(System.currentTimeMillis() - 100_000));
+ assertFalse(lockManager.isLockExpired(task));
+ }
+
+ // ------------------------------------------------------------------
lease stamping
+ // Every path that writes a lock must record the owner's lease with it,
and every path that
+ // clears a lock must clear the lease: a cleared owner with a leftover
lease (or the reverse)
+ // would make expiry decisions against a lock that no longer exists.
+
+ @Test
+ public void testParallelAcquireStampsLease() {
+ ScheduledTask task = TaskTestFixtures.baseTask("parallel-lease");
+ task.setAllowParallelExecution(true);
+ assertTrue(lockManager.acquireLock(task));
+ assertEquals(1000, task.getLockLeaseMillis(), "parallel marker must
record the owner's lease");
+ }
+
+ @Test
+ public void testInMemoryAcquireStampsLease() {
+ ScheduledTask task = TaskTestFixtures.baseTask("mem-lease");
+ task.setPersistent(false);
+ assertTrue(lockManager.acquireLock(task));
+ assertEquals(1000, task.getLockLeaseMillis(), "in-memory lock must
record the owner's lease");
+ }
+
+ @Test
+ public void testDistributedAcquireStampsLease() {
+ ScheduledTask task = TaskTestFixtures.baseTask("dist-lease");
+ task.setNextScheduledExecution(new Date(System.currentTimeMillis() -
10_000));
+ ScheduledTask latest = TaskTestFixtures.baseTask("dist-lease");
+ latest.setItemId(task.getItemId());
+ latest.setSystemMetadata("seq_no", 3L);
+ latest.setSystemMetadata("primary_term", 1L);
+ when(schedulerService.getTask(task.getItemId())).thenReturn(latest);
+
when(schedulerService.saveTaskWithRefresh(any(ScheduledTask.class))).thenReturn(true);
+
+ assertTrue(lockManager.acquireLock(task));
+ assertEquals(1000, task.getLockLeaseMillis(), "distributed lock must
record the owner's lease");
+ }
+
+ /**
+ * Renewal re-stamps the lease from the owner's CURRENT timeout, so a
runtime configuration
+ * change (ConfigAdmin update) propagates to the store within one renewal
interval instead of
+ * peers judging against a stale grant for the rest of the execution.
+ */
+ @Test
+ public void testRenewLockRestampsLeaseFromCurrentTimeout() {
+ ScheduledTask task = TaskTestFixtures.baseTask("renew-lease");
+ task.setLockOwner(NODE);
+ task.setLockDate(new Date());
+ task.setLockLeaseMillis(1000);
+
+ ScheduledTask storeView = TaskTestFixtures.baseTask("renew-lease");
+ storeView.setItemId(task.getItemId());
+ storeView.setLockOwner(NODE);
+ storeView.setLockDate(task.getLockDate());
+ storeView.setLockLeaseMillis(1000);
+ when(schedulerService.getTask(task.getItemId())).thenReturn(storeView);
+ when(schedulerService.saveTaskWithRefresh(storeView)).thenReturn(true);
+
+ lockManager.setLockTimeout(5000);
+ assertTrue(lockManager.renewLock(task));
+ assertEquals(5000, storeView.getLockLeaseMillis(), "store must carry
the current lease");
+ assertEquals(5000, task.getLockLeaseMillis(), "caller's view must be
synced to the current lease");
+ }
+
+ @Test
+ public void testReleaseLockClearsLease() {
+ ScheduledTask task = TaskTestFixtures.baseTask("release-lease");
+ task.setLockOwner(NODE);
+ task.setLockDate(new Date());
+ task.setLockLeaseMillis(1000);
+
+ ScheduledTask stored = TaskTestFixtures.baseTask("release-lease");
+ stored.setItemId(task.getItemId());
+ stored.setLockOwner(NODE);
+ stored.setLockDate(task.getLockDate());
+ stored.setLockLeaseMillis(1000);
+ when(schedulerService.getTask(eq(task.getItemId()),
eq(true))).thenReturn(stored);
+
+ assertTrue(lockManager.releaseLock(task));
+ assertEquals(0, task.getLockLeaseMillis(), "release must clear the
caller's lease");
+ assertEquals(0, stored.getLockLeaseMillis(), "release must clear the
persisted lease");
+ }
+
+ /**
+ * The recovery-enabling direction: a dead owner that granted itself a
SHORT lease is
+ * recoverable by an observer configured with a much longer timeout — the
observer must not
+ * impose its own, slower opinion on a lock whose owner promised to renew
far sooner.
+ */
+ @Test
+ public void testNonOwnerCanReleaseLockPastItsShortRecordedLease() {
+ lockManager.setLockTimeout(60_000); // observer is very patient by its
own config
+
+ ScheduledTask stored = TaskTestFixtures.baseTask("dead-short-lease");
+ stored.setLockOwner("dead-node");
+ stored.setLockDate(new Date(System.currentTimeMillis() - 2000));
+ stored.setLockLeaseMillis(500); // owner promised renewal every ~166ms
and is silent for 2s
+
+ ScheduledTask callerView =
TaskTestFixtures.baseTask("dead-short-lease");
+ callerView.setItemId(stored.getItemId());
+ callerView.setLockOwner("dead-node");
+ callerView.setLockDate(stored.getLockDate());
+ callerView.setLockLeaseMillis(500);
+ when(schedulerService.getTask(eq(callerView.getItemId()),
eq(true))).thenReturn(stored);
+
+ assertTrue(lockManager.isLockExpired(callerView),
+ "a lock silent past its own lease is expired even for a patient
observer");
+ assertTrue(lockManager.releaseLock(callerView),
+ "recovery must be able to clear a dead owner's expired-by-lease
lock");
+ assertNull(stored.getLockOwner());
+ assertEquals(0, stored.getLockLeaseMillis());
+ }
+
@Test
public void testAffinityBlocksBackupDuringPrimaryWindow() {
List<String> nodes = Arrays.asList("aaa-node", NODE, "zzz-node");
