Punctuation runs processor logic on a schedule instead of forwarding on every input record. A Punctuator supplies the callback through `Punctuator.punctuate(long timestamp)`, and `ProcessorContext.schedule` registers it with a punctuation type and interval. The returned `Cancellable` can cancel scheduled punctuation when an application needs that control. section 10.2.1
Choose the clock that should make the callback eligible to run:
- `STREAM_TIME` advances from record timestamps. Kafka Streams extracts the smallest timestamp across a task’s partition group and invokes eligible callbacks when scheduled time has been reached. Sporadic input means executions are not regular wall-clock intervals. section 10.2.2
- `WALL_CLOCK_TIME` follows Kafka Streams system time and is better for work that must happen despite idle input. It is best-effort: polling-cycle duration determines its granularity. section 10.2.2
Emit stock-performance records only after a scheduled review
Initialize a processor with its `ProcessorContext` and local key-value store, create a `StockPerformancePunctuator`, then schedule it every 10 seconds with `STREAM_TIME`. On each callback, iterate over stored performance entries and forward only those whose price or share differential meets the configured threshold. A callback can run without emitting a record, and `process` and `punctuate` do not run concurrently. section 10.2.1 section 10.2.4
Listing 10.9 Punctuation code
@Override
public void punctuate(long timestamp) {
try (KeyValueIterator<String, StockPerformance> performanceIterator
= keyValueStore.all()) { #1
while (performanceIterator.hasNext()) {
KeyValue<String, StockPerformance> keyValue =
performanceIterator.next();
String key = keyValue.key;
StockPerformance stockPerformance = keyValue.value;
if (stockPerformance != null) {
if (stockPerformance.getPriceDifferential() >=
differentialThreshold ||
stockPerformance.getShareDifferential() >=
differentialThreshold) { #2
context.forward(new Record<>(key, #3
stockPerformance,
timestamp));
}
}
}
}
}
Inspect the callback implementation that scans a state store, tests differential thresholds, and forwards qualifying records downstream. section 10.2.4
Punctuation also handles cleanup that input records alone cannot guarantee. A data-driven aggregation schedules wall-clock cleanup, scans its store for dangling aggregations, deletes expired entries, and forwards their final values. Track observed stream time carefully when cleanup decisions depend on timestamps, because out-of-order records must not move it backward. section 10.3
Test punctuation at two levels. Unit-test a Punctuator by mocking its `ProcessorContext` and state store, then verify forwarded records. For topology tests, provide input timestamps to advance stream time; for wall-clock punctuation, explicitly advance `TopologyTestDriver` wall-clock time. section 14.1.2 section 14.1.4
