Redpanda does a lot of things that are interesting to know about, but not necessarily bad or against-spec. For instance, g0 cycles are normal in the Kafka transactional model, and g1c is normal with wr-only edges at read-uncommitted but not with read-committed. This is a very ad-hoc attempt to encode that so that Jepsen’s valid/invalid results are somewhat meaningful.

Takes a test, and returns a set of keyword error types (e.g. :poll-skip) which this test considers allowable.

Builds up intermediate data structures used to understand a history. Options include:

:directory - Used for generating output files :ww-deps - Whether to perform write-write inference on the basis of log offsets.

Filters a history to just those operations around a given key and offset; trimming their mops to just those regions as well.

Filters a history to just those operations around a given key and value; trimming their mops to just those regions as well.

Clips a sequence to just those elements near a predicate. Takes a predicate, a range n, and a sequence xs. Returns the series of all x in xs such x is within n elements of some x’ matching predicate.

An assoc on vectors which allows you to assoc at arbitrary indexes, growing the vector as needed. When v is nil, constructs a fresh vector rather than a map.

Takes a test and a pair of an error type (e.g. :lost-write) and a seq of errors. Returns a pair of [type, {:count n, :errors …}], which tries to show the most interesting or severe errors without making the pretty-printer dump out two gigabytes of EDN.

Kafka transactions are supposed to offer ‘exactly once’ processing: a transaction using the subscribe workflow should be able to consume an offset and send something to an output queue, and if this transaction is successful, it should happen at most once. It’s not exactly clear to me how these semantics are supposed to work–it’s clearly not once per consumer group, because we routinely see dups with only one consumer group. As a fallback, we look for single consumer per process, which should DEFINITELY hold, but… appears not to.

We verify this property by looking at all committed transactions which performed a poll while subscribed (not assigned!) and keeping track of the number of times each key and value is polled. Yields a map of keys to values to consumed counts, wherever that count is more than one.

A generator which, if the test has :crash-clients? true, periodically emits an operation to crash a random client.

Finds a map of cycle names to cyclic anomalies in a partial analysis.

Turns a bifurcan integer map of Bifurcan sets, and converts it to a vector of Clojure sets.

Sometimes we wind up feeding absolutely huge plots to gnuplot, which chews up a lot of CPU time. We downsample these points, skipping points which are close in both x and y.

Takes a partial analysis and identifies cases where a single value appears at more than one offset in a key.

Takes an atom containing a map of keys to offsets. Constructs a generator which:

1. Checks the topic-partition state from the admin API

2. Crashes the client, to force a fresh one to be opened, just in case there’s broken state inside the client.

3. Assigns the new client to poll every key, and seeks to the beginning

4. Polls repeatedly

This process repeats every 10 seconds until polls have caught up to the offsets in the offsets atom.

First for vectors.

Takes a partial analysis and looks for aborted reads, where a known-failed write is nonetheless visible to a committed read. Returns a seq of error maps, or nil if none are found.

A combined Elle dependency graph between completion operations.

Takes a seq of distinct values, and returns a map of:

{:by-index A vector of the sequence :by-value A map of values to their indices in the vector.}

Takes a partial analysis and looks for cases where a single transaction contains:

{:skip A pair of poll values which read the same key and skip over some part of the log which we know should exist. :nonmonotonic A pair of poll values which contradict the log order, or repeat the same value.}

When a transaction’s rebalance log includes a key which would otherwise be involved in one of these violations, we don’t report it as an error: we assume that rebalances invalidate any assumption of monotonically advancing offsets.

A reducer for int-poll-skip+nonmonotonic-cases. Takes version orders, an op, a rebalanced-keys set, a transient vector of error maps and a key, values pair from (op-reads). Adds an error if we can find one in some key.

Takes a partial analysis and looks for cases where a single transaction contains a pair of sends to the same key which:

{:skip Skips over some indexes of the log :nonmonotonic Go backwards (or stay in the same place) in the log}

Takes CLI options (:sub-p) and a txn generator. Keeps track of the keys flowing through it, interspersing occasional :subscribe or :assign operations for recently seen keys.

Takes a key, a log for that key (a vector of offsets to sets of elements which were observed at that offset) and a history of ops relevant to that key. Constructs an XML structure visualizing all sends/polls of that log’s offsets.

Takes a log: a vector of sets of read values for each offset in a partition, possibly including nils. Returns a vector which takes indices (dense offsets) to sets of values whose last appearance was at that position.

Takes a log: a vector of sets of read values for each offset in a partition, possibly including nils. Returns a map which takes a value to the index where it first appeared.

Takes a partial analysis and looks for cases of lost write: where a write that we should have observed is somehow not observed. Of course we cannot expect to observe everything: for example, if we send a message to Redpanda at the end of a test, and don’t poll for it, there’s no chance of us seeing it at all! Or a poller could fall behind.

What we do instead is identify the highest read value for each key v_max, and then take the set of all values prior to it in the version order: surely, if we read v_max = 3, and the version order is 1 2 3 4, we should also have read 1 and 2.

It’s not quite this simple. If a message appears at multiple offsets, the version order will simply pick one for us, which leads to nondeterminism. If an offset has multiple messages, a successfully inserted message could appear nowhere in the version order.

To deal with this, we examine the raw logs for each key, and build two index structures. The first maps values to their earliest (index) appearance in the log: we use this to determine the highest index that must have been read. The second is a vector which maps indexes to sets of values whose last appearance in the log was at that index. We use this vector to identify which values ought to have been read.

Once we’ve derived the set of values we ought to have read for some key k, we run through each poll of k and cross off the values read. If there are any values left, they must be lost updates.

Takes an operation, a function f (:poll or :send), a key k, and a value v. Returns the index (0, 1, …) within that operation’s value which performed that poll or send, or nil if none could be found.

Takes a reads-by-type map and a (presumably :info) transaction which sent something. Returns true iff the transaction was :ok, or if it was :info and we can prove that some send from this transaction was read.

Takes a partial analysis and checks each process’s operations sequentially, looking for cases where a single process’s sends to a given key go backwards relative to the version order.

Nth for vectors, but returns nil instead of out-of-bounds.

Takes an operation (presumably, an OK or info one) and returns a map of keys to the highest offsets interacted with, either via send or poll, in that op.

Takes an operation and returns a map of keys to the highest offsets polled.

Takes an operation and returns a map of keys to the highest offsets sent.

Returns the thread which executed a given operation.

Takes an operation and returns that operation with its value trimmed so that any send/poll operations are constrained to just the given key, and values within n of the given offset. Returns nil if operation is not relevant.

Takes an operation and returns that operation with its value trimmed so that any send/poll operations are constrained to just the given key, and values within n of the given value. Returns nil if operation is not relevant.

Returns a map of keys to the sequence of all offset value pairs either written or read for that key; writes first.

Returns a map of keys to the sequence of all offsets read for that key.

Returns a map of keys to the sequence of all offset value pairs read for that key.

Returns a map of keys to the sequence of all values read for that key.

Takes an operation and a function which takes an offset-value pair. Returns a map of keys read by this operation to the sequence of (f offset value) read for that key.

We call op-reads a LOT. This takes a history and builds an efficient index, then returns a function (op-reads op) which works just like (op-reads op), but is memoized.

Returns a map of keys to the sequence of all offsets written to that key in an op.

Returns a map of keys to the sequence of all offset value pairs written to that key in an op.

Returns a map of keys to the sequence of all values written to that key in an op.

Takes an operation and a function which takes an offset-value pair. Returns a map of keys written by this operation to the sequence of (f offset value) sends for that key. Note that offset may be nil.

Quickly determine {:min-x, :max-x, :min-y, :max-y} from a series of x y points. Nil if there are no points.

Takes a test, a collection of realtime lag measurements, and options (e.g. those to checker/check). Plots a graph file (realtime-lag.png) in the store directory

Constructs realtime lag plots for all processes together, and then another broken out by process, and also by key.

Takes a test, a collection of unseen measurements, and options (e.g. those to checker/check). Plots a graph file (unseen.png) in the store directory.

Takes a partial analysis and checks each process’s operations sequentially, looking for cases where a single process either jumped backwards or skipped over some region of a topic-partition. Returns a task of a map:

{:nonmonotonic Cases where a process started polling at or before a previous operation last left off :skip Cases where two successive operations by a single process skipped over one or more values for some key.}

Per-process helper for poll-skip+nonmonotonic cases.

Wraps a generator. Keeps track of every offset that is successfully sent, and every offset that’s successfully polled. When there’s a key that has some offsets which were sent but not polled, we consider that unseen. This generator occasionally rewrites assign/subscribe operations to try and catch up to unseen keys.

Takes a partial analysis with a history and looks for a transaction which observed its own writes. Returns a vector of error maps, or nil if none are found.

This is legal in most DBs, but in Kafka’s model, sent values are supposed to be invisible to all pollers until their producing txn commits.

Takes a version order for a key and a value. Returns the previous value in the version order, or nil if either we don’t know v2’s index or v2 was the first value in the version order.

Takes a history and an op-reads fn, and builds a map of keys to values to vectors of completion operations which observed those that value.

Takes a history and an op-reads fn, and constructs a map of types (:ok, :info, :fail) to maps of keys to the set of all values which were read for that key. We use this to identify, for instance, the known-successful reads for some key as a part of finding lost updates.

Returns a seq of all operations which read the given key, and, optionally, read the given value.

Returns a seq of all operations which read the given key and offset.

Returns a seq of all operations which read the given key and value.

Takes a history and yields a series of maps of the form

{:process The process performing a poll :key The key being polled :time The time the read began, in nanos :lag The realtime lag of this key, in nanos.

The lag of a key k in a poll is the conservative estimate of how long it has been since the highest value in that poll was the final message in log k.

For instance, given:

{:time 1, :type :ok, :value [:send :x 0 :a]} {:time 2, :type :ok, :value [:poll {:x 0 :a}]}

The lag of this poll is zero, since we observed the most recent completed write to x. However, if we:

{:time 3, :type :ok, :value [:send :x 1 :b]} {:time 4, :type :invoke, :value :poll} {:time 5, :type :ok, :value [:poll {:x []}]}

The lag of this read is 4 - 3 = 1. By time 3, offset 1 must have existed for key x. However, the most recent offset we observed was 0, which could only have been the most recent offset up until the write of offset 1 at time 3. Since our read could have occurred as early as time 4, the lag is at least 1.

Might want to make this into actual lower upper ranges, rather than just the lower bound on lag, but conservative feels OK for starters.

Takes a test, an analysis, and for each key with certain errors renders an HTML timeline of how each operation perceived that key’s log.

Second for vectors.

Wraps a (jepsen.checker/stats) with a new checker that returns the same results, except it won’t return :valid? false if :crash or :debug-topic-partitions ops always crash. You might want to wrap your existing stats checker with this.

Takes a collection of maps, and removes their :type fields. Returns nil if none remain.

Takes a generator and tags operations as :f :poll or :send if they’re entirely comprised of send/polls.

Wraps a generator. Keeps track of every key that generator touches in the given atom, which is a map of keys to highest offsets seen.

Takes a list-append generator and rewrites its transactions to be :poll or :send k v micro-ops. Also adds a :keys field onto each operation, with a set of keys that txn would have interacted with; we use this to generate :subscribe ops later.

Takes a partial analysis and yields a series of maps like

{:time The time in nanoseconds :unseen A map of keys to the number of messages in that key which have been successfully acknowledged, but not polled by any client.}

The final map in the series includes a :messages key: a map of keys to sets of messages that were unseen.

Takes a history and a reads-by-type structure. Constructs a map of:

{:orders A map of keys to orders for that key. Each order is a map of: {:by-index A vector which maps indices to single values, in log order. :by-value A map of values to indices in the log. :log A vector which maps offsets to sets of values in log order.}

:errors A series of error maps describing any incompatible orders, where a single offset for a key maps to multiple values.}

Offsets are directly from Kafka. Indices are dense offsets, removing gaps in the log.

Note that we infer version orders from sends only when we can prove their effects were visible, but from all polls, including :info and :fail ones. Why? Because unlike a traditional transaction, where you shouldn’t trust reads in aborted txns, pollers in Kafka’s transaction design are always supposed to emit safe data regardless of whether the transaction commits or not.

Takes a logs object from version-orders and a micro-op, and integrates that micro-op’s information about offsets into the logs.

Updates a version orders log with the given offset and value.

Constructs a workload (a map with a generator, client, checker, etc) given an options map. Options are:

:crash-clients? If set, periodically emits a :crash operation which the client responds to with :info; this forces the client to be torn down and replaced by a fresh client.

:crash-client-interval How often, in seconds, to crash clients. Default is 30 seconds.

:sub-via A set of subscription methods: either #{:assign} or #{:subscribe}.

:txn? If set, generates transactions with multiple send/poll micro-operations.

:sub-p The probability that the generator emits an assign/subscribe op.

These options must also be present in the test map, because they are used by the checker, client, etc at various points. For your convenience, they are included in the workload map returned from this function; merging that map into your test should do the trick.

… plus those taken by jepsen.tests.cycle.append/test, e.g. :key-count, :min-txn-length, …

Takes a seq of realtime lag measurements, and finds the point with the highest lag.

Analyzes a history to extract write-read dependencies. T1 < T2 iff T1 writes some v to k and T2 reads k.

Takes a history and builds a map of keys to values to the completion operation which attempted to write that value.

Takes a history and constructs a map of types (:ok, :info, :fail) to maps of keys to the set of all values which were written for that key. We use this to identify, for instance, what all the known-failed writes were for a given key.

Returns a seq of all operations which wrote the given key, and, optionally, sent the given value.

Returns a seq of all operations which wrote the given key and offset.

Returns a seq of all operations which wrote the given key and value.

Analyzes a history to extract write-write dependencies. T1 < T2 iff T1 sends some v1 to k and T2 sends some v2 to k and v1 < v2 in the version order.