Throttle

Description

Limits the emission rate of events per group using a token bucket. For each group (defined by Key by), a bucket of Burst capacity tokens refills at one token every Time window / Number of events. An event is emitted immediately if a token is available, otherwise it is buffered and released when the next token refills. Events are never dropped and never reordered within a group. After an idle period up to Burst capacity events are released at once; the sustained rate stays at most Number of events per Time window. The default is 100 events per 1 second per group, burst 100.

The component declares no output variable and passes all scenario variables through. It is a rate shaper, not a mapping - emitted events are passed through unchanged and only their timing is affected.

Use cases

• Protecting a rate-limited downstream API, sink or database from bursts.
• Per-tenant or per-customer pacing, so one noisy key cannot consume the whole downstream budget.
• Flattening a replay or backfill spike so a slower consumer keeps up.
• Capping notification volume per recipient without losing any message.
• Spreading writes into an external system with a documented QPS budget.

Parameters and configuration

Name	Description
Key by	Expression assigning each event to a group; events sharing a key are throttled independently. Example: #input.customerId. Leave as '' to throttle the whole stream as one group. Events whose key is null are reported as errors and not emitted.
Number of events	Maximum number of events emitted per Time window, per group. Must be at least 1. Default 100.
Time window	Window over which Number of events events are released per group. Effective spacing is Time window / Number of events. Must not be negative. Default 1 second.
Burst capacity (advanced)	Maximum number of events released instantly after an idle period. Empty = Number of events. 1 = no burst (even spacing). Greater than Number of events allows larger spikes at the same average rate. Must be at least 1.
Time mode (advanced)	Processing time (wall clock, default) or Event time. In processing time the schedule follows the wall clock; in event time it follows each event's own timestamp and emissions happen as the watermark advances.

Number of events and Time window are only ever used as the ratio Time window / Number of events, so 100 per 1 second and 1 per 10 milliseconds describe the same sustained rate. They differ only in the default they imply for Burst capacity, which is Number of events when left empty.

Example

To emit at most 100 events per second per customer:

• Key by: #input.customerId
• Number of events: 100
• Time window: 1 second

Each customer's events are released at up to 100/s; bursts up to 100 pass immediately after a lull.

How it works

1. Each event is grouped by Key by.
2. The component schedules each event at max(now, TAT - tau), where TAT is the per-group theoretical arrival time and tau = (Burst capacity - 1) * (Time window / Number of events). This releases up to Burst capacity events immediately after idle, then paces the rest.
3. An event whose schedule falls on now is emitted straight away, provided the group has no backlog. Otherwise it is buffered under its fire time, and a timer (processing- or event-time, per Time mode) releases every event sharing that fire time together.

Marble example: one group, Number of events = 1, Time window = 5 seconds (so the spacing is 5s), Burst capacity = 3. Five events arrive at once at t=0.

in    t=0  e1 e2 e3 e4 e5
out   t=0  e1 e2 e3
      t=5  e4
      t=10 e5

• e1, e2 and e3 are within the burst tolerance (tau = 2 x 5s), so they go out immediately.
• e4 and e5 exhaust it and are paced one spacing apart.
• Had the five events arrived 5 seconds apart, none would have been buffered at all: at that rate the accumulator never gets ahead of real time, so the burst tolerance is never touched.

An eligible event still waits behind a backlog

Ordering within a group is preserved even when it costs latency. Emitting an event straight away requires two things: the token bucket has to allow it and the group's buffer has to be empty. An event that the bucket alone would let through is therefore still buffered while its group holds a backlog, and it leaves on the next timer instead of overtaking the events queued in front of it. This shows up mostly in event time, where a whole batch of events can be processed before any timer fires.

End of input and drain

When the scenario input ends (a bounded source finishing, or stop with drain), any still-buffered events are flushed in fire-time order so they are not lost. The remaining delay is not awaited - waiting on the task thread would stall checkpoints and job completion - so the configured rate is not honoured for that final flush and a downstream system can see a burst as the job finishes. In event time the same thing happens through the final watermark emitted at end of input.

Restarts

In event time the timeline is watermark-driven, so a restart replays it and the schedule resumes on its own.

In processing time the timers restored from the checkpoint would all be past due and would dump the whole backlog the moment the job resumes - exactly the burst the component exists to prevent. Instead each group's backlog is drained on restore, the pacing accumulator is cleared so that the outage looks like an idle period, and every buffered event is rescheduled from the moment the job comes back: up to Burst capacity events go out at once and the rest are paced again.

The cost is that a buffered event is delayed by the duration of the outage plus its position in the re-paced schedule. The backlog itself survives the restart and is still reported by throttle.bufferedEvents after the restore.

When to use Throttle instead of a similar component

• Delay shifts each event by a delay computed for that event, so the output has the same shape as the input, just later. Throttle changes the shape: it flattens bursts to a bounded rate, and the delay an event gets is whatever the backlog in front of it requires.
• Debounce collapses a burst of changes into a single settled emission and deliberately drops the transient ones. Throttle never drops - it spreads the same events over time.
• Deduplication with a #passedEventsCount < N filter condition caps how many events pass per key per window and discards the rest. Use it when exceeding the budget should shed traffic, and Throttle when every event has to be delivered eventually.
• A tumbling window reduces a burst to one aggregated record per window. Throttle keeps every record intact and only decides when each one leaves the node.

Additional considerations

• No back-pressure, unbounded buffering. If a group's input rate sustainedly exceeds its throttle rate, its buffer grows without bound. The component delays events; it does not drop them and it does not slow the source down. Size the rate against the sustained input rate rather than the average, and watch throttle.bufferedEvents for a backlog that never drains.
• Per-group state grows with key cardinality. The per-group pacing accumulator is retained for every key for the lifetime of the job, is not evicted for idle keys and has no TTL. State size therefore grows with the number of distinct keys ever seen, not with traffic: a key space of stable cardinality (customer ids, tenant ids, device ids) is fine, while an unbounded one (request ids, session ids) will grow the checkpoint without limit. Key by a bounded attribute.
• Buffered events sit in checkpointed state. Each buffered event is stored in keyed state under its fire time, with one timer per distinct fire time. A large backlog therefore makes checkpoints large and slow, so size the state backend for the worst-case backlog rather than for the steady state.
• Effective rate is capped at ~1000 events/s per group by millisecond timer granularity. Configurations requiring a sub-millisecond spacing are rejected at validation, with the error reported on Time window. To go above that rate, spread the traffic over more keys - each group has its own budget - rather than raising Number of events.
• The spacing is truncated to whole milliseconds, so the effective rate can exceed the configured one. 300 events per 1 second gives a spacing of 3 ms, which is 333 events/s rather than 300. Pick values whose ratio divides evenly (1000 per 1 second, 100 per 1 second, 1 per 10 milliseconds) when the exact rate matters.
• Event time assumes per-key ordering. In event time each event is scheduled against its own timestamp while the accumulator advances in arrival order, so a strongly out-of-order key can produce fire times that are already behind the watermark and are released at once, breaching the intended rate. Kafka partitioned by the throttle key maintains the needed ordering; processing time avoids the issue entirely.

Metrics

• throttle.bufferedEvents - number of events currently buffered (per node).
• throttle.delay - histogram of the delay scheduled for an event at the moment it arrives, in milliseconds. Events emitted immediately are recorded as 0, so the histogram covers all traffic through the node, not only the buffered part. The value is the delay decided on arrival: the extra hold that a processing-time restart adds by re-pacing the backlog is not reflected in it.

Both are reported through the standard Nussknacker metrics pipeline and tagged with nodeId and nodeName, so they can be narrowed down to a single Throttle node. They land in the same metrics backend as the built-in scenario metrics and can be plotted in Grafana - the shipped dashboard has no Throttle panel, so add one. See Nussknacker metrics for the metrics architecture and the fields each metric type reports.

Watch throttle.bufferedEvents to tell a node that is pacing a backlog from one that is passing everything straight through, and throttle.delay to see how long events are actually being held against the rate you configured.

throttle.bufferedEvents is only meaningful summed across subtasks. The counter is kept per subtask and checkpointed as evenly-redistributed operator state, while the buffers it counts live in keyed state, which is redistributed by key group. After a change of parallelism the two no longer line up: an individual subtask can report a value that never returns to zero, or a negative one, while the sum over all subtasks stays correct. Plot the sum, not a single subtask.

Configuration

The component is auto-loaded and needs no configuration. To change the default the Designer offers for Time mode, add a section for it in the model configuration:

components.throttle {
  timeMode = "ProcessingTime"   # or "EventTime"
}

The value is matched exactly, so it has to be spelled as above. This is a default for newly added nodes only - it does not change scenarios that are already deployed.