rbed

Resuming a stream

An application that reads a message stream (a Kafka topic, say) wants to stop and start again, for a deployment or after a crash, and continue as if it had never stopped: the same Events at the same timestamps with the same flags and Aggregates, each reported once. History restores only Ended Events. Three capabilities restore the rest (Events in progress, running durations, moving-average windows, rate(), age(), maxGap, latest(...) Aggregates):

  • Snapshot. detector.snapshot() returns the complete state as a byte[], and restore(bytes) on the builder continues from it, with nothing to replay. A restored Detector is indistinguishable from the original in everything observable. Take it between moments (never inside a listener callback) and store it with the position of the next message.
  • Replay Point. detector.replayPoint() is the earliest timestamp from which replaying the stream into a fresh Detector reproduces the state, when the Events that Ended before it are supplied as History. Nothing is stored but a timestamp and Ended Events. It is the way to continue when there is no usable Snapshot, and it bounds the replay after an older one.
  • Memory Horizon. detector.memoryHorizon() is the longest time window and the largest Sample count the Rules use. It sizes the topic's retention and the Snapshot interval. It is a configuration bound, not a guarantee: an Event in progress can reach further back. Only the Replay Point is exact.

A machine that overheats

A machine reports its temperature and load once a second, 120 messages. Overheat holds while the temperature is above 80 for 3 seconds, and is over when the load drops below 10, which can happen while the temperature is still high. Warm holds while the average of the last five temperatures is above 60, for 3 Samples. Repeated holds while more than one Overheat was Confirmed within the last minute.

The process dies after a message, and a new process takes over from the next one. Choose the stop and how the new process restarts, and compare its Events with those of a Detector that never stopped.

The live chart and its controls need JavaScript. The tables show the Events at the default parameters.

The site cannot run the default parameters: (JavaScript) TypeError: Cannot read properties of undefined (reading 'length')

import static io.github.romanhaendler.rbed.Conditions.*;

import io.github.romanhaendler.rbed.*;
import java.time.Duration;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.OptionalLong;

Channel temp = Channel.named("temp");
Channel load = Channel.named("load");
Rule overheat = Rule.named("Overheat").when(temp.above(80))
    .forDuration(Duration.ofSeconds(3)).endWhen(load.below(10));
Rule warm = Rule.named("Warm")
    .when(temp.averageOverSamples(5).above(60)).forSamples(3);
Rule repeated = Rule.named("Repeated")
    .when(count(overheat).within(Duration.ofSeconds(60)).above(1));
Rule[] rules = {overheat, warm, repeated};

List<Sample> topic = new ArrayList<>();   // the stream: one message a second
for (int s = 0; s < 120; s++) {
  topic.add(new Sample(s * 1_000L, Map.of(temp, temperature(s), load, loadAt(s))));
}

int stop = 45;   // the process dies after this many messages
List<Event> ended = new ArrayList<>();   // stored with each message
boolean[] replaying = {false};
EventListener listener = new EventListener() {
  @Override public void onEnded(Event e) {
    if (!replaying[0]) ended.add(e);
  }
};
Detector detector = Detector.builder().rules(rules).listener(listener).build();
for (int i = 0; i < stop; i++) {
  detector.push(topic.get(i).timestamp(), topic.get(i).values());
}
OptionalLong replayPoint = detector.replayPoint();   // stored with each message
byte[] snapshot = detector.snapshot();               // stored every so often
// the process dies here: detector is never closed

// restore the Snapshot and go on with the next message
Detector restarted = Detector.builder().rules(rules)
    .restore(snapshot).listener(listener).build();
for (int i = stop; i < topic.size(); i++) {
  restarted.push(topic.get(i).timestamp(), topic.get(i).values());
}
restarted.close();

// In the enclosing class:
static double temperature(int s) {
  return s < 10 ? 40 : s < 30 ? 90 : s < 40 ? 50 : s < 70 ? 85 : s < 90 ? 45 : s < 100 ? 88 : 45;
}

static double loadAt(int s) {
  return (s >= 20 && s < 35) || (s >= 60 && s < 80) || s >= 100 ? 5 : 50;
}

At the defaults, the process dies after message 45, at 44 s, while the second Overheat is running (Pending since 40 s, Confirmed at 43 s) and so is Warm (since 41 s). The uninterrupted Detector reports Overheat from 10 to 20 s and from 40 to 60 s, Warm from 12 to 33 s and from 41 to 73 s, Repeated from 43 to 74 s, then Overheat from 90 to 100 s, Warm from 91 to 103 s and Repeated from 93 to 104 s.

  • Snapshot and Replay Point report exactly these Events. The Replay Point at the stop is 36 s: Warm started at 41 s because the average of the Samples from 37 to 41 s rose above 60, and the replay has to see that it was not above 60 at 40 s, which needs the Samples from 36 s on (Overheat alone would need only 39 s, the last Sample on which it was seen not to hold). The Memory Horizon is the minute of count(...).within(...) and the five Samples of the average, 60 s and 5 Samples; the Replay Point is much closer in time.
  • History only knows the Events that Ended before the stop, but not the one in progress. The new Detector starts Overheat at 45 s with a Truncated start, and Repeated therefore starts at 48 s, not 43 s. Warm starts at 49 s, also Truncated: its average begins again with an empty window.
  • Naive, a fresh Detector, knows nothing: Overheat also starts at 45 s Truncated, and Warm at 49 s. It has lost the Overheat Confirmed at 13 s as well, so Repeated does not start at 48 s at all, and its second one, from 93 s, counts the Overheat of 45 s instead of 40 s and ends at 109 s, five seconds late.

Things to try

  • Move the stop to message 5, before anything happened: all four restarts equal the uninterrupted run, there is nothing to lose.
  • Move it to message 80, after the second Overheat: History only is exact too, since every Event has Ended. The naive restart loses the count of Events, so Repeated is missing at 93 s.
  • Move it to message 105, after the last Overheat: again nothing is left to lose.
  • Move it to message 100, while the third Overheat is running: History only fails again, and the Replay Point is 86 s: Overheat went Pending at 90 s, and Warm at 91 s on an average whose window starts at 86 s.

What to persist, and how to restart

After each message (or each batch), in one transaction with its side effects, or idempotently, store the position of the next message; for the Snapshot recipe the Snapshot and the position after it (every message, or every few seconds or minutes); for the replay recipe the Replay Point and the Ended Events (from onEnded), keyed by Rule name and start, so that storing one twice changes nothing.

  1. Restore and continue. Build the Detector with the same Rules and options and .restore(snapshot), seek to the position after the Snapshot and go on.
  2. Replay from the Replay Point. Build the Detector with .history(...) of the stored Ended Events that ended before the stored Replay Point, seek to the first message at or after the Replay Point and feed the messages from there. The replay reports Events that already happened, so suppress the listener's side effects until the replay reaches the last processed position.
  3. An older Snapshot plus the tail. Restore as in 1, then replay the messages from the Snapshot's position to the last processed one with side effects suppressed, then continue. Snapshots can be taken rarely; the Replay Point is not needed.

A Snapshot written by one runtime restores in the other: the Snapshot of the Java code on this page is the one the page's compiled copy of rbed writes in your browser, and the site's tests restore each in the other and check that both continue identically.

Restoring needs a Detector built with the same Rules (also names and order), watched Expressions, sampling rate and Retention Limit, and no history(...); anything else is rejected, so a deployment that changed a Rule falls back to the replay. The README has the full recipe with a runnable example, the preconditions and notes on Kafka.

Reference

Methods used: Channel.named, Rule.named, Rule.when, Rule.forDuration, Rule.endWhen, Expression.above, Expression.below, Conditions.count, Detector.builder, Detector.Builder.rules, Detector.Builder.listener, Detector.Builder.history, Detector.Builder.restore, Detector.Builder.build, EventListener.onEnded, Detector.push, Detector.replayPoint, Detector.snapshot, Detector.close.