Cluster-aware task scheduler

Unomi 3.1 introduces a first-class task scheduler for work that must run reliably on a single node or across a cluster: maintenance timers, request offload (for example profile-merge follow-up), and operator-visible retry/cancel.

It sits on the persistence-based clustering model — executor nodes and locks use the shared search index rather than Cellar or Hazelcast.

Capabilities at a glance

• Task types + executors — each logical taskType is handled by a registered TaskExecutor (core services or plugins)
• Persistent or in-memory — cluster-visible scheduledTask items in Elasticsearch/OpenSearch, or nonPersistent() node-local tasks (typical for cache refresh)
• One-shot and recurring — asOneShot(), or period + withFixedDelay() / withFixedRate(), with optional withInitialDelay
• Cluster coordination — executor nodes (scheduler.executorNode), cluster locks, optional runOnAllNodes(), parallel execution control via disallowParallelExecution()
• Lifecycle — statuses SCHEDULED, WAITING, RUNNING, COMPLETED, FAILED, CANCELLED, CRASHED
• Retries — withMaxRetries / withRetryDelay (auto); manual retry via REST/shell (optional failure-count reset)
• Dependencies — withDependencies(taskId…) so a task waits until prior work is COMPLETED
• Checkpoints / resume — long jobs save checkpointData / currentStep; operators resume CRASHED tasks when the executor implements canResume / resume
• Parameters — JSON-serializable map passed into the executor (include tenantId and run under executeAsTenant for tenant isolation)
• System tasks — asSystemTask() for jobs created at init that should be preserved across restarts
• Fluent API — schedulerService.newTask(type) builder; also createRecurringTask / createTask helpers
• Plugin registration — register/unregister TaskExecutor from OSGi DS activate / deactivate
• REST ops — list (filter by status/type, pagination), get, cancel, retry, resume (admin / tenant-admin); task creation stays in Java services/plugins
• Karaf shell — unomi:task-list|show|cancel|retry|purge|executor
• Built-in purge — periodic task-purge of old COMPLETED records (scheduler.purgeTaskEnabled, TTL days)
• Metrics — getMetric / getAllMetrics (completed, failed, crashed, running, lock conflicts, recovery attempts, …)
• Config — thread pool size, stable scheduler.nodeId, lock timeout, executor vs API-only node roles in org.apache.unomi.services.cfg

REST surface

Base path /cxs/tasks (roles ROLE_UNOMI_ADMIN / ROLE_UNOMI_TENANT_ADMIN) — monitoring and operations, not create:

• GET /cxs/tasks — paginated list; filters status=, type=, offset, limit
• GET /cxs/tasks/{id} — detail (parameters, checkpoint, errors)
• DELETE /cxs/tasks/{id} — cancel (HTTP 204; record kept as CANCELLED)
• POST /cxs/tasks/{id}/retry — optional ?resetFailureCount=true
• POST /cxs/tasks/{id}/resume — for CRASHED + checkpoint-capable executors

Karaf shell

unomi:task-list [-s STATUS] [-t TYPE] [--limit N]
unomi:task-show <id>
unomi:task-cancel <id>
unomi:task-retry <id> [--reset]
unomi:task-purge [--days N] [--force]
unomi:task-executor          # is this node an executor? nodeId?

Fluent scheduling (plugins / services)

schedulerService.newTask("cache-refresh-segment")
    .nonPersistent()
    .withPeriod(1, TimeUnit.SECONDS)
    .withFixedDelay()
    .disallowParallelExecution()
    .withSimpleExecutor(() -> refreshSegmentCache())
    .schedule();

schedulerService.newTask("export-tenant-snapshot")
    .withParameters(Map.of("tenantId", "acme"))
    .asOneShot()
    .withMaxRetries(3)
    .withRetryDelay(2, TimeUnit.MINUTES)
    .disallowParallelExecution()
    .withExecutor(myExportExecutor)
    .schedule();

Built-in task types you will see

• cache-refresh-* — reload segments, rules, property types, scopes, … into memory
• profile-purge
• segment-date-recalculation
• rules-statistics-refresh
• task-purge — purge old completed scheduler tasks
• merge-profiles-reassign-data — async session/event reassignment after profile merge (baseplugin)
• clusterNodeStatisticsUpdate / clusterStaleNodesCleanup — clustering maintenance

Async enrichment-style jobs can reuse the same model instead of ad-hoc threads — register an executor, schedule with the builder, operate via REST/shell.

Manual: scheduler Platform foundation Request tracing

← All posts