KAFKA-14491: [7/N] Enforce strict grace period for versioned stores by vcrfxia · Pull Request #13243 · apache/kafka

vcrfxia · 2023-02-14T01:06:39Z

The RocksDB-based implementation for versioned key-value stores introduced in #13188 has a well-defined "history retention" parameter which specifies how far back in time (relative to the current observed stream time) reads may take place, but there is no well-defined equivalent (aka "grace period") for how far back in time writes will be accepted. Instead, there is an implicit grace period whereby the store accepts all writes which affect valid reads. This doesn't quite work, though, because it requires infinite tombstone retention when the latest value for a particular key is a tombstone -- if the latest value for a key is a very old tombstone, we can’t expire it because if there’s an even older non-null put to store later, then without the tombstone we’ll accept this write as the latest value for the key, even though it isn't.

In light of this, this PR changes the versioned store semantics to define an explicit "grace period" property. (KIP-889 has been updated accordingly.) For now, grace period will always be equal to the history retention, though in the future we can introduce a new KIP to expose options to configure grace period separately.

Committer Checklist (excluded from commit message)

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

vcrfxia · 2023-02-14T01:17:34Z

+    }
+
+    @Test
+    public void shouldAllowZeroHistoryRetention() {


Added this extra test in response to a previous PR review comment.

This is an interesting edge case in that if history retention = grace period = 0, then we don't actually need the segments store because grace period = 0 means we don't need to store tombstones. (Even if a tombstone is the latest value for a given key, the store will never accept earlier writes so the store doesn't need to keep the tombstone after clearing the current value for the key.)

Is it worth it to add extra code to remove the segments store in this case? My instinct says no because this case does not seem very practical. For a user to use grace period = 0 requires that they are confident that all records within a partition, even across keys, are produced in ascending (technically, non-decreasing) timestamp order. I'm not sure how many use cases meet this criterion.

mjsax

Overall LGTM. Few minor questions.

One more though: should we add verification about the "droppedRecordSensor" into all unit tests that drop records? We could add it to this PR or do a follow up (whatever you prefer). -- We sometimes have bugs that we don't maintain metrics correctly, so getting some testing might be a good thing.

mjsax · 2023-02-14T23:24:21Z

        // on the specific workload and the value of the "segment interval" parameter.
        for (final ConsumerRecord<byte[], byte[]> record : records) {
+            if (record.timestamp() < streamTimeForRestore - gracePeriod) {
+                // record is older than grace period and was therefore never written to the store


If it was never written to the store, if should also not be in the changelog topic?

This might still be useful if we read from the input topic for a KTable I guess? But we might want to update the JavaDoc for to mention this case?

If it was never written to the store, if should also not be in the changelog topic?

Ideally, but unfortunately no. Only the inner layer (RocksDBVersionedStore) contains logic for deciding when grace period has elapsed and a call to put() should return without updating the store. The changelogging layer wrapped around this inner layer does not know about grace period, nor do any of the other outer layers. The changelogging layer does call put() before calling log(), but because put() has no return type, it does not convey information about whether an update was actually made or if put() simply returned without doing anything. So, the changelogging layer calls log() in either case.

This is the existing behavior for window stores, and what I had planned to replicate for versioned stores as well. If we don't want this, we could:

update put() to return a boolean, indicating whether the update was actually performed, or

track observed stream time and grace period at an outer store layer, in order to not call log() at the changelogging layer if it's not needed.

I don't particularly like either option. Curious to hear your thoughts.

Thanks. Makes sense.

I think it's ok to leave it as-is for now. But could you maybe file a Jira ticket (with all the glory details) for tracking? Might be worth to do some follow up work later to change it (but not worth to delay the KIP implemenation at this point).

Sounds good. Here's the ticket: https://issues.apache.org/jira/browse/KAFKA-14723

mjsax · 2023-02-14T23:27:07Z

-        // advance stream time to the max timestamp in the batch
+        // copy the observed stream time, for use in deciding whether to drop records during restore,
+        // when records have exceeded the store's grace period.
+        long streamTimeForRestore = observedStreamTime;


Wondering if this would be correct?

If we have st = 100, grace=10 and we do put(k,v,95) the put is correct. If we restore at st=110, the would still need to keep k,v and not drop it, even if it's timestamp 95 is now "too old"?

Yeah this logic is pretty nuanced. (I tried to clarify in the comments but evidently not successfully.)

The doPut() method is not responsible for deciding when a put is too old (according to grace period); that check happens before doPut() is called. Inside the doPut() method, however, observedStreamTime is still used to decide when old records have fallen out of history retention. If a record has fallen out of history retention, then we don't need to keep it in the store, and therefore doPut() returns.

In this restore logic here, streamTimeForRestore is used to perform the grace period check. It would be incorrect to advance streamTimeForRestore at once for the entire batch, for the reason you gave above. In your example, we do still want to call doPut() for the record with ts=95. Assuming that is the first record in the restore batch, then streamTimeForRestore=100 so ts=95 and we call doPut() as we should. Only once we reach the later records in the restore batch will streamTimeForRestore be advanced past 100.

OTOH, observedStreamTime can be advanced to the end of the batch right away. This allows us to optimize situations where, for example, a record near the beginning of the restore batch which we would put into the store would be immediately expired (based on history retention) by the end of the restore batch, and therefore we can skip putting it in inside doPut(). Here's an example:

stream time is 50 at the start of the restore batch

segment interval is 25

stream time will be 100 by the end of the restore batch

restore batch contains a record (k, v, 50) and also (k, v, 60).

During restore when we see (k, v, 50), we have to put it into the store (it's the latest value for the key so far). Then when we see (k, v, 60), we also have to put it into the store (it's the new latest value) but we do NOT have to move (k, v, 50) into a segment store, because the segment that it would be moved into will be expired by the end of the restore process.

Here's another example: exact same as above, but the restore batch contains (k, v, 60) before (k, v, 50), instead of after. When we see (k, v, 60) we have to put it into the store. When we see (k, v, 50), we still call doPut() because it's not expired based on grace period, but doPut() will see that it is expired based on history retention (using observedStreamTime=100, the value it will be by the end of the restore batch) and therefore doPut()returns without inserting into the store.

I guess the question is, what is the value of observedStreamTime when we start the restore? Are you saying it's -1 and we basically "reply" observedStreamTime during restore? I guess I got confused with "streamTime" that is tracked by KS runtime and preserved across restarts; but the store does not use it (IIRC), but rather tracks its own time, right?

Maybe best to update some variable names? In the end, we do a "real reply" of stream-time for "grace period", and we apply an optimization for "history retention" by looking ahead (to the end of the batch) -> endOfBatchStreamTime. -- I guess follow up work (independent for this KIP) might be, to actually make use of KS runtime streamTime instead of tracking inside the store, and thus won't need observedStreamTime any longer, as we could look ahead to the "end-of-restore stream-time" (not just "end-of batch").

I guess the question is, what is the value of observedStreamTime when we start the restore? Are you saying it's -1 and we basically "replay" observedStreamTime during restore?

Yes, that's exactly right. observedStreamTime is tracked locally per store. It is initialized to -1 and only updated on put() or during restore. (This is the same as the existing behavior for window stores today.)

Maybe best to update some variable names?

Are you proposing that doPut() takes stream time as a parameter, so that during normal put() operation we pass observedStreamTime and during restore we pass endOfBatchStreamTime, which means we can rename streamTimeForRestore to be observedStreamTime instead? This SGTM, just want to check whether that's also what you have in mind, since we removed a number of parameters from doPut() in a previous PR revision in order to keep the parameter list small.

I guess follow up work (independent for this KIP) might be, to actually make use of KS runtime streamTime instead of tracking inside the store, and thus won't need observedStreamTime any longer, as we could look ahead to the "end-of-restore stream-time" (not just "end-of batch").

What's the scope of the "streamTime" which is tracked by the KS runtime? Is it per-task? Per-processor? Global? I'm wondering how this would work in situations with multiple partitions, or with multiple processors where some processors are expected to see new data earlier than other (downstream) processors.

I guess we'd also need to implement the change from your other comment about not writing records which are expired (based on grace period) into the changelog topic first before we can make this change, otherwise we would not have a way to determine during restore whether records are expired or not.

Are you proposing that doPut() takes stream time as a parameter, so that during normal put() operation we pass observedStreamTime and during restore we pass endOfBatchStreamTime, which means we can rename streamTimeForRestore to be observedStreamTime instead?

Went ahead and made this update in the latest commit. Can revise if it's not what you had envisioned.

Did not have a concrete proposal. Should be fine I guess.

Currently, streamTime is tracked per task (based on input records over all partitions). And yes, there is all kind of tricky things that you call out. Even if we have a filter() downstream processors see only a subset of data and their "internal stream-time (if they have any)" could be different (ie lagging). Caching has a similar effect.

There is a proposal to let KS track streamTime per processor, too.

Bottom line: it's complicated and need proper design and a KIP by itself...

mjsax · 2023-02-14T23:30:07Z

-                        expiredRecordSensor.get().record(1.0d, context.currentSystemTimeMs());
-                        LOG.warn("Skipping record for expired put.");
-                    }
+                    // the record being inserted does not affect version history. discard and return.


Not sure if I can follow. Why did we record this in the sensor first, but not any longer?

Same below (2x).

With the changes in this PR, it is only possible to hit this case during restore now. Previously, we passed Optional.empty() for the expiredRecordSensor anyway, because we don't want to call the sensor during restore. So I've simplified the code by removing it entirely.

The reason it is not possible to hit this case during non-restore is because doPut() is not called if the record being put is older than grace period, and history retention is always at least as large as grace period. (See my comment above for why it is still possible to hit this case during restore.)

mjsax · 2023-02-14T23:33:44Z

+        final List<DataRecord> records = new ArrayList<>();
+        records.add(new DataRecord("k", "v", HISTORY_RETENTION + 10));
+        records.add(new DataRecord("k1", "v1", HISTORY_RETENTION + 10 - GRACE_PERIOD)); // grace period has not elapsed
+        records.add(new DataRecord("k2", "v2", HISTORY_RETENTION + 9 - GRACE_PERIOD)); // grace period has elapsed, so this record should not be restored


Cf comment above. The question seems to be "when" the original put() happened with regard to stream-time?

That's correct. This test case uses the same data as shouldNotPutExpired() above. This third record is expired even during normal put operations.

Thx. After you pointed out the "store hierarchy" in your above reply and that the record would go into the changelog, the test makes sense.

mjsax · 2023-02-14T23:37:42Z

+        // query in past (history retention expired) returns null
+        verifyTimestampedGetNullFromStore("k", BASE_TIMESTAMP + 1);
+
+        // put in past (grace period expired) does not update the store


Should we also test put-in-past-for-existing record?

Sure, I can add that. I was worried that the test case was already getting a bit long :)

Hm, just realized it's not possible to add this case in a meaningful way. Suppose observed stream time is t and we put-in-past for an existing key at time t-1. We cannot query for the value of the key at time t-1 because that is outside history retention. And if we query for the latest value of the key, then we'll get the record at time t regardless of whether the put at time t-1 was properly rejected or not.

We'd have to query the inner store in order to perform this check, which feels like overkill. WDYT?

Was just an idea. Not a big deal to not have the test.

Just see you added the test. Does not hurt to keep it. (We should not write test base on knowing how the implemenation works, but rather treat it as a "black box").

mjsax

Overall LGTM. Few minor questions.

One more though: should we add verification about the "droppedRecordSensor" into all unit tests that drop records? We could add it to this PR or do a follow up (whatever you prefer). -- We sometimes have bugs that we don't maintain metrics correctly, so getting some testing might be a good thing.

vcrfxia

Thanks @mjsax for your review! Responded to your questions inline. Will push a new commit shortly.

One more thought: should we add verification about the "droppedRecordSensor" into all unit tests that drop records?

Let me look into the best way to test this. If it doesn't require a bunch of changes for how to expose the metrics/sensor so that the test can access them, then I will add it into this PR.

vcrfxia · 2023-02-15T00:35:57Z

-        // advance stream time to the max timestamp in the batch
+        // copy the observed stream time, for use in deciding whether to drop records during restore,
+        // when records have exceeded the store's grace period.
+        long streamTimeForRestore = observedStreamTime;


Yeah this logic is pretty nuanced. (I tried to clarify in the comments but evidently not successfully.)

The doPut() method is not responsible for deciding when a put is too old (according to grace period); that check happens before doPut() is called. Inside the doPut() method, however, observedStreamTime is still used to decide when old records have fallen out of history retention. If a record has fallen out of history retention, then we don't need to keep it in the store, and therefore doPut() returns.

In this restore logic here, streamTimeForRestore is used to perform the grace period check. It would be incorrect to advance streamTimeForRestore at once for the entire batch, for the reason you gave above. In your example, we do still want to call doPut() for the record with ts=95. Assuming that is the first record in the restore batch, then streamTimeForRestore=100 so ts=95 and we call doPut() as we should. Only once we reach the later records in the restore batch will streamTimeForRestore be advanced past 100.

OTOH, observedStreamTime can be advanced to the end of the batch right away. This allows us to optimize situations where, for example, a record near the beginning of the restore batch which we would put into the store would be immediately expired (based on history retention) by the end of the restore batch, and therefore we can skip putting it in inside doPut(). Here's an example:

stream time is 50 at the start of the restore batch

segment interval is 25

stream time will be 100 by the end of the restore batch

restore batch contains a record (k, v, 50) and also (k, v, 60).

During restore when we see (k, v, 50), we have to put it into the store (it's the latest value for the key so far). Then when we see (k, v, 60), we also have to put it into the store (it's the new latest value) but we do NOT have to move (k, v, 50) into a segment store, because the segment that it would be moved into will be expired by the end of the restore process.

Here's another example: exact same as above, but the restore batch contains (k, v, 60) before (k, v, 50), instead of after. When we see (k, v, 60) we have to put it into the store. When we see (k, v, 50), we still call doPut() because it's not expired based on grace period, but doPut() will see that it is expired based on history retention (using observedStreamTime=100, the value it will be by the end of the restore batch) and therefore doPut()returns without inserting into the store.

vcrfxia · 2023-02-15T00:40:05Z

        // on the specific workload and the value of the "segment interval" parameter.
        for (final ConsumerRecord<byte[], byte[]> record : records) {
+            if (record.timestamp() < streamTimeForRestore - gracePeriod) {
+                // record is older than grace period and was therefore never written to the store


If it was never written to the store, if should also not be in the changelog topic?

Ideally, but unfortunately no. Only the inner layer (RocksDBVersionedStore) contains logic for deciding when grace period has elapsed and a call to put() should return without updating the store. The changelogging layer wrapped around this inner layer does not know about grace period, nor do any of the other outer layers. The changelogging layer does call put() before calling log(), but because put() has no return type, it does not convey information about whether an update was actually made or if put() simply returned without doing anything. So, the changelogging layer calls log() in either case.

This is the existing behavior for window stores, and what I had planned to replicate for versioned stores as well. If we don't want this, we could:

update put() to return a boolean, indicating whether the update was actually performed, or

track observed stream time and grace period at an outer store layer, in order to not call log() at the changelogging layer if it's not needed.

I don't particularly like either option. Curious to hear your thoughts.

vcrfxia · 2023-02-15T00:42:27Z

-                        expiredRecordSensor.get().record(1.0d, context.currentSystemTimeMs());
-                        LOG.warn("Skipping record for expired put.");
-                    }
+                    // the record being inserted does not affect version history. discard and return.


With the changes in this PR, it is only possible to hit this case during restore now. Previously, we passed Optional.empty() for the expiredRecordSensor anyway, because we don't want to call the sensor during restore. So I've simplified the code by removing it entirely.

The reason it is not possible to hit this case during non-restore is because doPut() is not called if the record being put is older than grace period, and history retention is always at least as large as grace period. (See my comment above for why it is still possible to hit this case during restore.)

vcrfxia · 2023-02-15T00:43:22Z

+        final List<DataRecord> records = new ArrayList<>();
+        records.add(new DataRecord("k", "v", HISTORY_RETENTION + 10));
+        records.add(new DataRecord("k1", "v1", HISTORY_RETENTION + 10 - GRACE_PERIOD)); // grace period has not elapsed
+        records.add(new DataRecord("k2", "v2", HISTORY_RETENTION + 9 - GRACE_PERIOD)); // grace period has elapsed, so this record should not be restored


That's correct. This test case uses the same data as shouldNotPutExpired() above. This third record is expired even during normal put operations.

vcrfxia · 2023-02-15T00:44:50Z

+        // query in past (history retention expired) returns null
+        verifyTimestampedGetNullFromStore("k", BASE_TIMESTAMP + 1);
+
+        // put in past (grace period expired) does not update the store


Sure, I can add that. I was worried that the test case was already getting a bit long :)

vcrfxia · 2023-02-15T20:14:04Z

One more thought: should we add verification about the "droppedRecordSensor" into all unit tests that drop records?

Included this test update in the latest commit. I believe I've addressed/responded to all outstanding comments with the latest commit.

mjsax · 2023-02-15T20:49:06Z

+     * history retention, even though history retention is always at least the grace period,
+     * during restore because restore advances {@code observedStreamTime} to the largest timestamp
+     * in the entire restore batch at the beginning of restore, in order to optimize for not
+     * putting records into the store which will have expired by the end of the restore.


Thanks for adding this! Great addition!

mjsax

LGTM. Can merge after Jenkins passed.

mjsax added streams kip Requires or implements a KIP labels Feb 14, 2023

enforce strict grace period = history retention

ad011db

vcrfxia force-pushed the kip-889-versioned-store-strict-grace-period branch from 7442470 to ad011db Compare February 14, 2023 01:08

vcrfxia mentioned this pull request Feb 14, 2023

KAFKA-14491: [5/N] Basic operations for RocksDB versioned store #13188

Merged

3 tasks

vcrfxia commented Feb 14, 2023

View reviewed changes

add warn log on expired get

f98429a

mjsax reviewed Feb 14, 2023

View reviewed changes

vcrfxia commented Feb 15, 2023

View reviewed changes

review feedback

fb76487

mjsax reviewed Feb 15, 2023

View reviewed changes

mjsax approved these changes Feb 15, 2023

View reviewed changes

mjsax merged commit bfeef29 into apache:trunk Feb 16, 2023

vcrfxia deleted the kip-889-versioned-store-strict-grace-period branch February 16, 2023 16:12

Uh oh!

Conversation

vcrfxia commented Feb 14, 2023 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Committer Checklist (excluded from commit message)

Uh oh!

Choose a reason for hiding this comment

Uh oh!

mjsax left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

mjsax left a comment

Choose a reason for hiding this comment

Uh oh!

vcrfxia left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

vcrfxia commented Feb 15, 2023

Uh oh!

Choose a reason for hiding this comment

Uh oh!

mjsax left a comment

Choose a reason for hiding this comment

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

vcrfxia commented Feb 14, 2023 •

edited

Loading