KAFKA-14491: [16/N] Add recovery logic for store inconsistency due to failed write

[vcrfxia]

The RocksDB-based implementation of versioned stores introduced in #13188 consists of a "latest value store" and separate (logical) "segments stores." A single put operation may need to modify multiple (two) segments, or both a segment and the latest value store, which opens the possibility to store inconsistencies if the first write succeeds while the later one fails. When this happens, Streams will error out, but the store still needs to be able to recover upon restart. This PR adds the necessary repair logic into RocksDBVersionedStore to effectively undo the earlier failed write when a store inconsistency is encountered.

Committer Checklist (excluded from commit message)

• Verify design and implementation
• Verify test coverage and CI build status
• Verify documentation (including upgrade notes)

[mjsax]

Did not look into the test yet, as I don't fully understand the actually code yet.

[mjsax]

Think we can improve this existing comment?

// Detected inconsistency edge case for partial update:
// We always insert base on `validTo`, and we would only have a "double write"
// if an existing record moves from a newer segment into an older segment
// (because its `validTo` timestamp changes due to the newly put record).
// For this case, we first move the existing record from the newer to the older segment and
// afterwards update the newer segment with the new record.
// If only the first part, ie, moving the existing record into the older segment succeeded,
// we end up with a "valid interval" overlap (and corrupted older segment) that we need to resolve:
//  - older segments latest record: [oldValue, old-validFrom, updated-validTo)
//    (this in incorrect because the dual write was not completed)
//  - newer segments oldest record: [oldValue, old-validFrom, old-validTo)
//    (this is correct, as it was never modified)
//
// We have the current older segment at hand, and need to truncate the partial write,
// ie, remove [oldValue, old-validFrom, updated-validTo) from it.

Is this correct?

[vcrfxia]

Yep, that's spot on. I will update the comment with your suggestions.

After a partial write failure is encountered, when the store next resumes processing then the store should first reprocess records which it has already processed previously (if EOS is not enabled), including the one during which the failure occurred, before processing any newly seen records. In this case, we would only expect a single record to be truncated (and to be truncated in full) as part of the recovery process -- as you've outlined above. The recovery code is more general, though, in that it also supports truncating multiple records and/or partial records, in case my above understanding about the types of failures which can occur is not accurate or changes in the future.

[mjsax]

The recovery code is more general, though, in that it also supports truncating multiple records and/or partial records, in case my above understanding about the types of failures which can occur is not accurate or changes in the future.

Yes, that is what I figured -- wondering if it would be safe to make it more generic or not? -- The order in which we process records is not 100% deterministic (we could have some s1.merge(s2).toTable() for example and read from two partitions).

Or in the "worst" case, somebody might write a non-deterministic custom Processor...

Just want to double check your thoughts on this.

[vcrfxia]

I actually think the examples you gave are great examples for why we need this logic to be more generic -- if those are "valid" cases which could be encountered during processing/re-processing, then we need the store to be able to properly handle them after encountering a failure.

The cleanup logic is safe because the only type of failure we can have is duplicated data, i.e., two segments (or one segment and the latest value store) contain overlapping records. When this happens, we know it is safe to truncate from the older segment because the data being truncated is also present in the newer segment.

[vcrfxia]

Updated the comment in the latest commit.

[mjsax]

Not sure if I understand? find() is inclusive, right? It would return a record with validFrom == timestamp as search result? JavaDocs say "not exceeding":

Finds the latest record in this segment row with (validFrom) timestamp not exceeding the
provided timestamp bound.

And the code inside find() says:

if (currTimestamp <= timestamp) {
    // found result

[vcrfxia]

Yes, find() is inclusive. That's also what "not exceeding" means (if the timestamps are equal, then neither exceeds the other), so I don't think there's a contradiction here?

[mjsax]

Well, if find() is inclusive, why would we need to bump the index by one? the index should be correct already?

[vcrfxia]

Suppose the older segment has records [validFrom=10, validTo=20) and [validFrom=20, validTo=30). Records are stored reverse-sorted by timestamp so [validFrom=20, validTo=30) has index 0 and [validFrom=10, validTo=20) has index 1. If we want to truncate to timestamp 20, calling find() will return index 0, but the number of (full) records to remove is 1 -- the entire record with index 0 must be removed.

In contrast, if we are truncating to timestamp 25, calling find() will also return 0. In this case, the number of records to remove is 0. No incrementing is needed here. Does that help?

[mjsax]

Not sure if I understand this condition?

[vcrfxia]

This catches the situation in which the last record of the segment being inserted into has validFrom=5 and validTo=10, and insertAsLatest() has been called with timestamp 8 (more generally, any timestamp larger than 5 and smaller than 10). In this situation, all we have to do to recover (before performing the insert) is to update validTo for the existing record from 10 to 8.

As noted in a comment above, I don't actually expect this situation to be hit but I think it is good to be defensive about these additional edge cases in this recovery logic, as it is difficult to anticipate the types of failures which may be encountered.

[mjsax]

Well, but if we want to make it generic, could it not be that we actually have two record [5,10)[10,15) and try to insert at 8? Or would we never call insertAsLatest() for this case?

[vcrfxia]

The logic added in this PR does handle this case. (It's not "expected" that this case will come up, but it is possible in weird edge cases such as the examples that you gave above.) There are unit tests for it as well.

[mjsax]

But if we handle this case, don't we incorrectly drop [10,15)? The intention if my example was to say that [5,10) was the partial write, but [10,15) was a clean write later (due to re-ordering) that did not detect the corruption and went through cleanly, and now inserting 8 would incorrectly purge [10,15)?

Or maybe this case could never happen? -- Just want to double check.

[vcrfxia]

That can never happen -- when a single put() call requires writing to multiple segments, it always writes to the latest record in the older segment and the earliest record in the newer segment. If [5,10) is a partial write, then that means it was written as the latest record to an older segment, and that the newer segment already has that same record timestamp 5 in it as its earliest record (which was supposed to be replaced with a new record except that second write failed). In that case, any subsequent write with timestamp greater than 5 will stop in the newer segment because the newer segment still has minTimestamp = 5.

[mjsax]

Don't we know, that we only need to remove the oldest entry at index 0? For what case would we need to remove more than one entry/version ?

[vcrfxia]

See above. I also can't think of situations in which we'd need to recovery by doing anything other than removing the current last record version in full, but I think it is good to be defensive. The current understanding that this is the only recovery scenario which is needed is based on a lot of factors (the fact that this recovery is only performed on insertAsLatest(), usage of insertAsLatest() from the store implementation, etc) which could potentially change in the future. I think it'd be good to future-proof this recovery logic in the event that something changes in the future, even though we do not expect that to be the case.

[mjsax]

but I think it is good to be defensive

In general yes, but only if we are sure we don't corrupt data. Are we sure about this?

[vcrfxia]

As the code stands today, it's safe yeah. The safety relies on some properties of the current usage of this class, though, such as the fact that insertAsLatest() is only called in certain cases, that we always insert into the older segment first when two segments must both be updated, etc. If these change in the future then it could become unsafe. So I guess the "future-proofing" that I intended with these more generic updates creates its own risk for un-future-proof changes as well.

Maybe the question then is what we think is worse -- if we don't allow generic truncation but instead require that only one full record can be truncated (which is what we expect to happen if there is a failure and then the same record(s) are re-processed right afterward), then the store will throw IllegalStateException when the assumption is violated. Users will have to delete local instances of the store and restore from changelog to recover. Alternatively, if we keep the generic changes, then that is not necessary (and is safe for now), but something could change in the future to make it unsafe (unlikely but possible).

I see the merits of both. Maybe it's better to start with the stricter requirement (do not support generic truncation) first and relax the requirement if users report hitting IllegalStateException. WDYT?

[mjsax]

That's a tricky one. Maybe @guozhangwang @vvcephei @cadonna can comment?

[vvcephei]

Agreed; it's a tricky question. It seems like either way, we need to clearly document/enforce the contract so that future generations can have a reasonably good chance of not breaking the system.

Ideally, there would be at one test verifying the contract is followed (eg, the order of inserts into the underlying segments), so that when someone breaks it, they'll also break the test, which can have a name and comment explaining the contract. The next-best thing would be to structure the code in a way that's pretty self-explanatory, though a sequence of refactors can destroy this property without any specific step being obviously wrong.

I don't think I like the idea of just throwing an IllegalStateException in production code. It puts the user in a really bad position where they've upgraded to a new version that is now crashing in production, and they have to decide whether it's safe to downgrade or whether they need to wait for a fix, and all the while they're incurring downtime. Note that the IllegalState condition may take a while to manifest; imagine what you'd be going through if your app started crashing with this error a month after your last upgrade.

OTOH, we do have recoverable exceptions in Streams, where you can signal to the runtime that the task is corrupted, triggering the purge-and-rebuild-from-changelog automatically. Perhaps that, coupled with an ERROR log telling the user to file a bug ticket, would be the best path forward.

[guozhangwang]

I'm also a bit leaning away from throwing an IllegalStateException to forbid such a case, instead, I think have a test case may be well sufficient as we do: admittedly we cannot just enforce this property at the callee side, but have to rely on the caller's behavior, but with a test case on the outer-side classes that covers the callers should cover the case if caller ever change to mal-behave.

[vcrfxia]

Thanks for the discussion, everyone! I chatted with Matthias and we've decided to not throw IllegalStateException or a recoverable exception, and instead log an extra warn message if more (or less) than a single record is truncated as part of this recovery logic, which would indicate either nondeterministic processing or that something has changed in the code to violate our current understanding that only one record should be truncated for recovery.

As Guozhang noted it's quite difficult to write tests to ensure that all the various existing properties that guarantee safety of this truncation today are preserved going forward as there are a bunch of them which all mutually interact. We have pretty thorough unit test coverage on the outer classes already. The only extra testing I can think to add would be tests which explicitly simulate partial failed updates and verify recovery. Might be overkill but I can do that in a follow-up PR if we think it's worth it.

[mjsax]

I think this is "strictly" not correct (we lost the original validTo / next_timestamp what was overwritten by moving a record into this segment in the partial write); but it also does not matter, because we actually have a put() at hand that will update next_timestamp right after we finished the cleanup again, right? (not sure if we should document this with a comment?)

[vcrfxia]

but it also does not matter, because we actually have a put() at hand that will update next_timestamp right after we finished the cleanup again

Yeah that's correct. My intention with updating nextTimestamp here even though it will be immediately updated again once we return from this method (back to insertAsLatest()) is so that the state of the segment is always "consistent," which makes the code easier to reason about IMO. Another way to think about it is that when an inconsistency is detected from insertAsLatest(), we respond by truncating all contents in this segment up to timestamp (the timestamp of the record which is being inserted). This truncation involves updating nextTimestamp for the segment to be timestamp.

Do you have thoughts on what additional comments (and where) would help clarify this? I had hoped that the method name truncateRecordsToTimestamp and this comment stating that we are updating nextTimestamp to timestamp would be sufficient, but it appears not :)

[mjsax]

Yeah -- not 100% sure -- guess my idea was just to avoid potential confusion if somebody read this code in the future (but also always tricky to not confuse readers with too much detail). Maybe something like:

// update nextTimestamp as part of truncation -- this a surrogate value that just fixed inconsistency without being actually correct; but we have a pending put() in place that will set the correct value later 

Just an idea.

[vcrfxia]

I opted to add a javadoc to the method -- LMK whether you think it clears up the confusion here.

I think it might be a bit too granular to add the explanation into this line itself because there's also another place in the method (the first if-condition were we replace the entire segment with a degenerate one if the truncation timestamp is less than or equal to the minTimestamp of the segment) where readers are likely to have the same question if it's not already clear that nextTimestamp will be updated to the truncation timestamp.

[vcrfxia]

The build failed with an error that looks unrelated to this PR:

[2023-03-31T00:06:22.083Z] FAILURE: Build failed with an exception.
[2023-03-31T00:06:22.083Z] 
[2023-03-31T00:06:22.083Z] * What went wrong:
[2023-03-31T00:06:22.083Z] A problem was found with the configuration of task ':rat' (type 'RatTask').
[2023-03-31T00:06:22.083Z]   - Gradle detected a problem with the following location: '/home/jenkins/jenkins-agent/workspace/Kafka_kafka-pr_PR-13364'.
[2023-03-31T00:06:22.083Z]     
[2023-03-31T00:06:22.083Z]     Reason: Task ':rat' uses this output of task ':clients:processTestMessages' without declaring an explicit or implicit dependency. This can lead to incorrect results being produced, depending on what order the tasks are executed.

@mjsax would you mind triggering a new build in case it's a flake?