KAFKA-14491: [6/N] Support restoring RocksDB versioned store from changelog#13189
Conversation
0ee4bd9 to
4f38854
Compare
There was a problem hiding this comment.
Heh, I was hoping to ask you about this myself :)
Currently the code restores an entire batch at a time, by reading the relevant values into memory and then writing at all once at the end. If we were able to detect (or catch) a potential out of memory in this process, we could flush the write buffer partway through the batch to skirt the issue.
From reading online, most sources do not recommend trying to catch OutOfMemoryException even though other sources say that if you have a graceful way to recover (which we do in this case) then it could be fine. Another option is we could set a limit on the amount of memory that we want to be available to the restore write buffer, similar to what's done for the Streams ThreadCache today, and flush whenever we exceed the limit (before we hit an actual OOM). This is probably the better/safer option, but figuring out a good default value will be tricky, and we'd probably also want to expose it as a config for advanced users / users with atypical setups. WDYT?
There was a problem hiding this comment.
We don't try to handle OutOfMemoryException so far, thus, I think it's ok to not start (and if we want to start, make a plan and do it everywhere).
There was a problem hiding this comment.
Because of the way that the versioned store implementation stores multiple records (for the same key) together in a single RocksDB "segment" entry, the chance that the new versioned store implementation hits an OOM could be significantly higher than for existing stores -- restoring a single changelog entry could require loading multiple records into memory. How high this memory amplification will be is very much dependent on the specific workload and the value of the segment interval parameter.
In light of this, it feels like there's a case to be made that we should handle OOM for versioned stores, but if you don't feel it's urgent I'm happy to defer.
There was a problem hiding this comment.
At least no reason to block this PR. \cc @guozhangwang WDYT about this question?
There was a problem hiding this comment.
Yeah, we were aware of the risk of OOM since during restoration we need to potentially buffer some records, and made a decision to live with that during the design phase. So I think it’s still okay to not capture OOM explicitly. Admitting the likelihood would be much larger now.
| observedStreamTime = Math.max(observedStreamTime, record.timestamp()); | ||
| } | ||
|
|
||
| final VersionedStoreClient<?> restoreClient = restoreWriteBuffer.getClient(); |
There was a problem hiding this comment.
The VersionedStoreSegment implementation used by the restore client (WriteBufferSegmentWithDbFallback) is currently private to the write buffer class, since the RocksDBVersionedStore doesn't care what the type is; all the outer class needs are the methods provided by the RocksDBVersionedStoreClient interface itself. So I've left the type out in order to avoid polluting the outer class with extra info it doesn't need.
There was a problem hiding this comment.
I see -- fair enough (in general, I am a fan on strong typing whenever possible, but it's ok I guess).
| /** | ||
| * @return client for writing to (and reading from) the write buffer | ||
| */ | ||
| VersionedStoreClient<?> getClient() { |
mjsax
left a comment
mjsax
left a comment
There was a problem hiding this comment.
Couple of nits. (Did not look into testing code yet.)
|
Thanks for your review, @mjsax ! Responded to your comments inline and also pushed a commit just now containing the requested changes. |
dc708c8 to
f9e9181
Compare
3 participants
(This PR is stacked on #13188. Only the last commit (
add restore) needs to be reviewed as part of this PR.)This PR builds on the new RocksDB-based versioned store implementation (see KIP-889) introduced in #13188 by adding code for restoring from changelog. The changelog topic format is the same as for regular timestamped key-value stores: record keys, values, and timestamps are stored in the Kafka message key, value, and timestamp, respectively. The code for actually writing to this changelog will come in a follow-up PR.
Committer Checklist (excluded from commit message)