Materialize: Preserve sink Kafka topic names

[awang]

It is very hard to make use of sinked data if the topic name can change, since most consumers would expect that to be constant.

Proposed change:
Update sinks so the topic does not change on crash-restart, and they support “exactly-once” semantics (duplicates are not seen by downstream consumers)

also see: https://github.com/MaterializeInc/materialize/issues/1728

[umanwizard]

Background

By default, Materialize saves only catalog metadata - enough information to restart computations from the beginning. To avoid re-writing an entire second copy of the data into a sinks, we currently take the approach of appending a random nonce to sink topic names on creation. Thus, if Materialize is re-started, the sinked results of its computations will be written, from the beginning, into fresh Kafka topics.

This causes a variety of operational issues; customers typically ask for us to help resolve these issues by writing to "the same" Kafka sink on restart. However, we would like to do this without sacrificing correctness; i.e., either without introducing duplicates into the stream, or if we must have duplicates, doing so in such a way that they can be deduplicated easily by downstream clients.

Product requirements

One possibly guiding request is for outage recovery - if the Materialize host goes down or needs to be restarted for whatever reason, we would like to be able to bring it back up without causing disruption to downstream consumers of the sinks; in particular, without having to manually reconfigure them with whatever the new topic name is and forcing them to restart their own workflows.

(There may be other reasons people want this as well, which might have their own set of requirements ...)

Possible solutions

1. Re-ingest the entire old sink on startup, run the computation again, and subtract the old sink from the new computation. That way, what is written out will only be the new data. This could in theory work but would require O(n) startup time in the size of the historical output. Thus, the only way to really make this feasible would be to ask people to turn on compaction of sinks.

2. Check what was _last_ written to the sink, to avoid writing it again

For example: My consistency topic says timestamp t_0 is complete, I also have three messages "foo", "bar", and "quux" at t_0+1. Don't sink anything until the frontier has passed t_0, and also apply ("foo", -1), ("bar", -1), ("quux", -1) to the sinked collection to avoid duplicating those. There is probably a significant amount of messy logic involved, especially since we (currently) don't necessarily write sink outputs in order of timestamp.

1. Write in a format that allows duplicates

A variant of 2. that accepts some duplicates in the un-committed timestamps would make things simpler. The "CDCv2" format Frank invented seems perfect for this - it is an envelope on top of Avro that encodes progress messages inline in streams in such a way that it's possible to correctly interpret streams with arbitrarily duplicated and reordered messages. However, we would have to either teach customers how to interpret this format, or just write the logic for them (maybe a simple "basic CDCv2 reader" library would be not too hard to write. The hard stuff has already been done in differential-dataflow, anyway).

Repeatable timestamps

Currently in real-time sources, timestamps are assigned arbitrarily, which will make possible solutions 2. or 3. above not work correctly, so we would need to fix that... (will elaborate further on this tomorrow).

[umanwizard]

@ruchirK, @frankmcsherry and @elindsey , I will continue writing more stuff about this tomorrow. Also, please add anything you think I've missed or forgotten about in the solution space.

[rjnn]

Another option: Remember (durably) what was written to each sink. This means that when writing to sinks, we'd have to write down some metadata somewhere. This could be a "sink metadata" Kafka topic, where we write down what timestamp, offset, we've committed per sink topic/partition.

[umanwizard]

@rjnn We basically already do this, with the consistency topic. However - as I described in option 2 - you need to write some logic to deal with un-committed data. E.g., if you have committed everything at timestamp t_0, but then you write a few messages to timestamp t_1, then crash before finishing t_1 and writing "t_1 is done" to the consistency topic, then on restart you need to avoid writing anything <= t_0, _and_ any of the partial t_1 stuff that you wrote.

[umanwizard]

Durable timestamps

Background

All known Materialize users are using what we call "real-time sources" in which the assignment of timestamps (and therefore, consistency boundaries) is arbitrary and based on wall-clock time. Thus, RT sources lack what one might call "durability"; that is, when one restarts a computation from scratch, it will not necessarily give the same answers, because timestamps may have changed.

Furthermore, different instantiations of the _same_ source might have _different_ timestamps, which makes the situation even worse. For example, if src is an RT source,

CREATE MATERIALIZED VIEW mv1 AS SELECT * FROM src;
CREATE MATERIALIZED VIEW mv2 AS SELECT * FROM src;
SELECT * FROM mv1 UNION ALL mv2;

might return rows with multiplicity 1, and SELECT * FROM mv1 EXCEPT ALL mv2 might be nonempty. Both of these results are impossible in standard SQL/relational algebra.

Why it matters

A lot of reasons, but for this particular task, anything we do requires detecting duplicate messages, which will be basically impossible if the new messages after restart have totally different timestamps from the "same" messages written before restart.

Desirable properties

From this discussion emerges a set of properties that we would like Materialize to have:

1. Standard SQL queries must not give impossible results. This should be true regardless of the type of source used. If a type of source is unable to fulfill this property, we should not support it at all.
2. _If possible_, sources should be consistent through time. That is; if the underlying data in the sources has not changed, and Materialize is restarted, the result of any query should be possible given any previously observed result. Since this property requires some level of persistence of timestamps, and not all users will need it - it is probably OK for it to be opt-in only.

Possible solutions

Whenever we accept data for a given source and timestamp it, we should remember that binding of a timestamp to a set of Kafka offsets, and require all future reads of the same source to use it. IIRC, timestamp.rs is probably already set up to do this, but we stopped... for some reason?

This would give us property (1).

In order to also get property (2), there are a few possibilities. One is: whenever we make a timestamping decision, we could persist it before letting it flow to the rest of the system. In fact, this is what we used to do - I'm not sure why we stopped. The disadvantages include causing permanent O(n) growth of the catalog. Perhaps it would be better to write these to some Kafka topic, but then you would have to wait for that to be committed (for whatever notion of "committed" Kafka has) before allowing the data to leave the source and flow to the rest of the system.

Another option is to only store _one_ piece of data per partition: the latest timestamp (call it t) , and the latest offset x bound to it. Then, on restart, timestamp _all_ data up to and including x with timestamp t. After having read through x, continue timestamping based on wall-clock time as usual.

This is essentially equivalent to having compacted everything through t. It would _not_ give us property (2) for arbitrary AS OF queries, since all the distinctions before time t would have been lost. But, that should be fine, we already discard these distinctions when we do compaction. However, it _would_ give us property (2) for simple, non-AS OF queries: after restart, any new query would incorporate all the previous data and therefore be one possible continuation of it. The disadvantage of this is that the entire input (up to x) would need to be ingested before any queries can return - but that is probably fine (and in fact the more correct behavior) for a large class of users.

[rjnn]

@rjnn We basically already do this, with the consistency topic. However - as I described in option 2 - you need to write some logic to deal with un-committed data. E.g., if you have committed everything at timestamp t_0, but then you write a few messages to timestamp t_1, then crash before finishing t_1 and writing "t_1 is done" to the consistency topic, then on restart you need to avoid writing anything <= t_0, and any of the partial t_1 stuff that you wrote.

Kafka has a commit protocol to do all of this atomically.

[umanwizard]

@rjnn Oh, nice, that could indeed make "option 2" somewhat simpler.

[krishmanoh2]

The other issue here is the current syntax is opposite to WYSIWYG. Could we keep the same topic name as in "CREATE SINK" rather than using it as a prefix?

CREATE SINK "materialize"."public"."sink_top_buy_categories" FROM "materialize"."public"."mv_top_buy_categories" INTO KAFKA BROKER '192.168.1.254' TOPIC 'sink_top_buy_categories' WITH ("consistency" = true) FORMAT AVRO USING CONFLUENT SCHEMA REGISTRY 'http://192.168.1.254:8081' WITH SNAPSHOT AS OF "now"();

krmanoha=> select * from mz_kafka_sinks;
 sink_id |                            topic                            
---------+-------------------------------------------------------------
 u53     | sink_top_buy_categories-u53-1605273997-9258380046558767833

[umanwizard]

@krishmanoh2 Indeed - once we come up with a way to preserve the topic name, we can probably just use the raw sink name, rather than appending a nonce.

[elindsey]

Getting myself organized before I start on this - this is more like a concrete restatement of what Brennan's said with more specifics.

Goals:

• A host configured with realtime consistency and Kafka sinks can crash or be manually restarted and 'pick up where it left off' with the same topic name.
• Minimal changes to consumers (ie. no requirement on CDCv2 in the sink topic)

Non-goals:

• Supporting sinks other than Kafka. Eventually we will want to do this, but it would likely mean requiring CDCv2. Given the prevalence of Kafka, there's value in doing a non-CDCv2 version specific to that sink type.
• Multipartition sinks. We don't currently support them, this work continues not supporting them.
• BYO consistency is entirely out of scope.
• Reingesting our own sinks as a source.
• Minimizing state accumulation. O(n) growth of the catalog is acceptable for v1.
• Spinning up a clean host with no accumulated state and having it continue appending to an existing topic. For now, persistent local storage like a reused EBS volume will be required for host migrations.

Description

Mz changes:

1. Minted timestamps must match for all instantiations of a given source. Requires dusting off the dead code in timestamp.rs and changing sources to get their timestamps from the coord instead of locally.
2. Minted timestamps should be persisted and reused across restarts. Requires adding persistence, detection of persisted state, and reuse to timestamp.rs. Timestamper must also signal the sink that we're in reuse mode.
3. Writing to sinks should be idempotent. This is a small configuration change and needs to be done regardless.
4. Sink should update the consistency topic in lockstep with the result topic and only after a timestamp has been closed. Requires batching completed timestamps into a Kafka transaction.
5. On startup, sink should read the latest complete timestamp from the consistency topic and drop any writes coming through the dataflow until caught up.

Customer changes:

This will require customers' kafka consumers to be in read-committed mode. This is the default on many clients (such as all those based on rdkafka), but not for primary main Java client.

Alternatives

The best alternative is a similar scheme but using CDCv2 to remove the need for Kafka transactions. This is likely the best route when supporting other sinks that don't offer transactional support, but it does require us building out a small CDCv2 client in a few languages for customer ease of use. While a good option and something we will probably do in the future, given Kafka's prevalence it's worthwhile doing something specific for that sink type that imposes minimal format restrictions.

Most other alternatives are merely refinements or iterative improvements of this proposal.

Testing

Specific areas that will need careful testing:

• Coordination between the timestamper and sink: the timestamper must be in charge of deciding whether or not we reuse an existing topic. If the timestamper has no persisted state but the topic exists and the sink is in reuse mode, then we'll end up writing bogus data. This should instead either be an error on startup or result in a fallback to the nonce write path.
• Need to understand interaction with Kafka source topic compression
• Kafka transactions will incur some write amplification and should be sized 'as big as possible without incurring unnecessary latency.' That plus the fact that sinking timestamps only once completed will result in more buffering and less precise control over batching semantics means that we will need to revisit sink tuning.

[elindsey]

Brief update: from the 'MZ Changes' list above, 3/4 are done and I'm starting on 5 soon; Ruchir's likely taking 1/2.

In the intervening time we had a long tangent discussing possible alternatives to 1/2 (timestamp persistence via the timestamper thread), but ultimately settled on the same design that was laid out here.

There is one additional customer requirement: we do need to curb the O(N) disk growth of timestamp persistence, and fairly soon. For the people currently interested it's okay to treat all historical data as a single "catch up" timestamp on restart, so the current plan is (roughly) to have the timestamper write down minted offset/timestamp pairs, have the sink record published timestamps, and periodically condense that to only the most recent published offset/timestamp pair.

[frankmcsherry]

and periodically condense that to only the most recent published offset/timestamp pair.

This seems sane. For which use cases is this not appropriate?

[elindsey]

This seems sane. For which use cases is this not appropriate?

Talked over slack awhile back, but for posterity - I don't (currently) know of any use cases where this isn't appropriate, but I had been hoping to avoid it for v1 to get the basic version working end to end sooner. There's one quirk with ts condensing too: if there's still a large amount of historical data (ie. kafka topics aren't pruned), then it may land us in a state where we repeatedly oom on startup since the ts/offset granularity acts as a pacing mechanism today.

Down to four concrete tasks now:

1. Sink needs to read the consistency topic on startup and hold back publishing up to the most recently commited ts
2. 'exactly once' needs to be wired through the config parameters, disable the nonce'd topic name generation, require a consistency topic, and not attempt to recreate the topic each time
3. ts persistence/condensing, and resurrection of the unified timestamper (Ruchir)
4. Kafka txn calls need to be moved to a separate thread so they don't block the workers. _may_ be able to get away with not doing this for the initial version just to make sure it's all functionally correct, but will definitely need to be done in the future.

[awang]

Update: we've begun QA on two of the major PRs: https://github.com/MaterializeInc/materialize/pull/6532 and https://github.com/MaterializeInc/materialize/pull/6866

[ruchirK]

This has merged with experimental support