Multithreading in Action Kits

Java

Each action runs as a single instance with one consumer thread that pulls work from the queue. That consumer can process up to N items concurrently, where N is the action's configured concurrency. This is decoupled from the number of instances and pods: raising concurrency adds in-memory workers, not threads or Valkey connections, so you can scale an action's throughput without scaling the deployment to extra pods.

Concurrency is bounded per action by a gate; in-flight work for an action never exceeds its limit. The default concurrency, when none is configured, is 1.

Actions must be thread-safe. A single action instance is shared across all concurrent invocations, so raising concurrency above 1 runs multiple invocations against the same object simultaneously. Actions must not keep per-invocation state in instance fields — see Actions Must Be Thread-Safe. Default concurrency is 1, so an action is only invoked concurrently once you raise its limit.

Configuring Concurrency at Runtime

Per-action concurrency can be changed at runtime with no restart. A change takes effect on running plugins within a few seconds — growing takes effect immediately, shrinking drains in-flight work first.

deltafi action concurrency list                                          # all actions and current concurrency
deltafi action concurrency set org.deltafi.core.action.egress.RestEgress 5
deltafi action concurrency reset org.deltafi.core.action.egress.RestEgress

list reports each action's effective concurrency, a Source column showing where the value comes from — default (the system default of 1), config (the plugin's declared actions.actionThreads value), or override (a runtime value set with set) — and a Tunable column. Only actions whose kit honors runtime changes (this Java action kit) are tunable; actions on older kits or other languages appear in list but set rejects them, and they are omitted from set/reset shell completion.

reset reverts an action to its declared bootstrap value — the plugin's actions.actionThreads setting if it has one, otherwise 1. It does not force the value to 1; to pin an action below its declared default, use set with the value you want.

Runtime overrides persist across restarts and are captured in system snapshots.

Bootstrap Default (application.yaml)

To set an action's starting concurrency before any runtime override is applied, add an entry under the actions namespace in your application.yaml:

actions:
  actionThreads:
    org.deltafi.core.action.FilterEgressAction: 2

The effective concurrency is the runtime override if one is set, otherwise this bootstrap value, otherwise 1. Any value is clamped to actions.maxConcurrency (default 64).

Threading inside an action

Actions must be thread-safe (see Actions Must Be Thread-Safe for the why). The framework already runs many DeltaFiles through an action in parallel, so you rarely need to manage threads yourself — raising the action's concurrency is the supported way to scale throughput.

Do

• Keep per-DeltaFile state in local variables and method arguments — never in shared/instance state.
• Make any collaborator you store and reuse thread-safe. Most HTTP clients and connection pools are; objects like a non-thread-safe date formatter or XML parser are not — create those per invocation or use a thread-safe variant.
• Prefer raising the action's configured concurrency over hand-rolling parallelism inside a single invocation.

Don't

• Don't spawn unbounded threads or a fresh thread pool per invocation. Your threads multiply by the action's concurrency (N concurrent invocations × your threads) and can exhaust the pod. If one DeltaFile genuinely needs internal parallelism, use a small bounded executor and shut it down before the method returns.
• Don't stash a value in shared state "to pass it between methods" — pass it as a parameter instead.
• Don't assume invocations run in order or that one can see another's state; they are independent and may overlap. Note that internal threads you start also bypass the per-action concurrency limit and its backpressure, making it easy to overwhelm a downstream system.

Go

The Go action kit supports configurable worker counts per action. Set the worker count globally or per action.

Global Worker Count

Set the ACTION_THREADS environment variable in your Dockerfile:

ENV ACTION_THREADS=2

Per-Action Worker Count

Set an environment variable using the pattern ACTIONS_ACTIONTHREADS_{UPPERCASED_ACTION_NAME}, where dots and hyphens in the action name are replaced with underscores:

ENV ACTIONS_ACTIONTHREADS_ORG_DELTAFI_MY_PLUGIN_MYTRANSFORMACTION=4

Or set it programmatically before calling RunPlugin:

plugin.SetWorkerCount("org.myorg.my-plugin.MyTransformAction", 4)

By default, if an action does not have a specific thread count configuration, it will use 1 worker goroutine.

C++

The C++ action kit supports the same configurable worker counts as Go.

Global Worker Count

Set the ACTION_THREADS environment variable in your Dockerfile:

ENV ACTION_THREADS=2

Per-Action Worker Count

Set an environment variable using the pattern ACTIONS_ACTIONTHREADS_{UPPERCASED_ACTION_NAME}, where dots and hyphens in the action name are replaced with underscores:

ENV ACTIONS_ACTIONTHREADS_ORG_DELTAFI_MY_PLUGIN_MYTRANSFORMACTION=4

Or set it programmatically:

plugin.set_worker_count("org.myorg.my-plugin.MyTransformAction", 4);

By default, if an action does not have a specific thread count configuration, it will use 1 worker thread.