Skip to content

KAFKA-14491: [7/N] Enforce strict grace period for versioned stores #13243

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
mjsax merged 3 commits into from
Feb 16, 2023
Merged

KAFKA-14491: [7/N] Enforce strict grace period for versioned stores #13243

Show file tree
Hide file tree
Changes from 1 commit
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
ad011db
enforce strict grace period = history retention
vcrfxia Feb 13, 2023
f98429a
add warn log on expired get
vcrfxia Feb 14, 2023
fb76487
review feedback
vcrfxia Feb 15, 2023
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
13 changes: 13 additions & 0 deletions streams/src/main/java/org/apache/kafka/streams/state/VersionedKeyValueStore.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -30,6 +30,11 @@
* to return accurate results for calls to {@link #get(Object, long)} where the provided timestamp
* bound is within history retention of the current observed stream time. (Queries with timestamp
* bound older than the specified history retention are considered invalid.)
* <p>
* The store's "history retention" also doubles as its "grace period," which determines how far
* back in time writes to the store will be accepted. A versioned store will not accept writes
* (inserts, updates, or deletions) if the timestamp associated with the write is older than the
* current observed stream time by more than the grace period.
*
* @param <K> The key type
* @param <V> The value type
Expand All @@ -38,6 +43,10 @@ public interface VersionedKeyValueStore<K, V> extends StateStore {

/**
* Add a new record version associated with the specified key and timestamp.
* <p>
* If the timestamp associated with the new record version is older than the store's
* grace period (i.e., history retention) relative to the current observed stream time,
* then the record will not be added.
*
* @param key The key
* @param value The value, it can be {@code null}. {@code null} is interpreted as a delete.
Expand All @@ -52,6 +61,10 @@ public interface VersionedKeyValueStore<K, V> extends StateStore {
* <p>
* This operation is semantically equivalent to {@link #get(Object, long) #get(key, timestamp)}
* followed by {@link #put(Object, Object, long) #put(key, null, timestamp)}.
* <p>
* If the timestamp associated with this deletion is older than the store's grace period
* (i.e., history retention) relative to the current observed stream time, then the deletion
* will not be performed.
*
* @param key The key
* @param timestamp The timestamp for this delete
Expand Down
57 changes: 35 additions & 22 deletions streams/src/main/java/org/apache/kafka/streams/state/internals/RocksDBVersionedStore.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -22,7 +22,6 @@
import java.nio.ByteBuffer;
import java.util.Collection;
import java.util.List;
import java.util.Optional;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.common.metrics.Sensor;
import org.apache.kafka.common.utils.Bytes;
Expand Down Expand Up @@ -82,6 +81,7 @@ public class RocksDBVersionedStore implements VersionedKeyValueStore<Bytes, byte

private final String name;
private final long historyRetention;
private final long gracePeriod;
private final RocksDBMetricsRecorder metricsRecorder;

private final RocksDBStore latestValueStore;
Expand All @@ -101,6 +101,10 @@ public class RocksDBVersionedStore implements VersionedKeyValueStore<Bytes, byte
RocksDBVersionedStore(final String name, final String metricsScope, final long historyRetention, final long segmentInterval) {
this.name = name;
this.historyRetention = historyRetention;
// history retention doubles as grace period for now. could be nice to allow users to
// configure the two separately in the future. if/when we do, we should enforce that
// history retention >= grace period, for sound semantics.
this.gracePeriod = historyRetention;
this.metricsRecorder = new RocksDBMetricsRecorder(metricsScope, name);
this.latestValueStore = new RocksDBStore(latestValueStoreName(name), name, metricsRecorder);
this.segmentStores = new LogicalKeyValueSegments(segmentsStoreName(name), name, historyRetention, segmentInterval, metricsRecorder);
Expand All @@ -111,9 +115,14 @@ public class RocksDBVersionedStore implements VersionedKeyValueStore<Bytes, byte
@Override
public void put(final Bytes key, final byte[] value, final long timestamp) {

if (timestamp < observedStreamTime - gracePeriod) {
expiredRecordSensor.record(1.0d, context.currentSystemTimeMs());
LOG.warn("Skipping record for expired put.");
return;
}

doPut(
versionedStoreClient,
Optional.of(expiredRecordSensor),
key,
value,
timestamp
Expand Down Expand Up @@ -283,7 +292,12 @@ public void init(final StateStoreContext context, final StateStore root) {

// VisibleForTesting
void restoreBatch(final Collection<ConsumerRecord<byte[], byte[]>> records) {
// 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;

@mjsax mjsax Feb 14, 2023

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

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"?

@vcrfxia vcrfxia Feb 15, 2023

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

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.

@mjsax mjsax Feb 15, 2023

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

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").

@vcrfxia vcrfxia Feb 15, 2023

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

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.

@vcrfxia vcrfxia Feb 15, 2023

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

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.

@mjsax mjsax Feb 15, 2023

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

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...

// advance stream time to the max timestamp in the batch, in order to speed up restore
// by not bothering to read from/write to segments which will have expired by the time
// the restoration process is complete.
for (final ConsumerRecord<byte[], byte[]> record : records) {
observedStreamTime = Math.max(observedStreamTime, record.timestamp());
}
Expand All @@ -297,6 +311,12 @@ void restoreBatch(final Collection<ConsumerRecord<byte[], byte[]>> records) {
// 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.
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

@mjsax mjsax Feb 14, 2023

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

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?

@vcrfxia vcrfxia Feb 15, 2023

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

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.

@mjsax mjsax Feb 15, 2023

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

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).

@vcrfxia vcrfxia Feb 15, 2023

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

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

continue;
}
streamTimeForRestore = Math.max(streamTimeForRestore, record.timestamp());

ChangelogRecordDeserializationHelper.applyChecksAndUpdatePosition(
record,
consistencyEnabled,
Expand All @@ -306,7 +326,6 @@ void restoreBatch(final Collection<ConsumerRecord<byte[], byte[]>> records) {
// put records to write buffer
doPut(
restoreClient,
Optional.empty(),
new Bytes(record.key()),
record.value(),
record.timestamp()
Expand Down Expand Up @@ -440,7 +459,6 @@ public void writeLatestValues(final WriteBatch batch) throws RocksDBException {

private <T extends VersionedStoreSegment> void doPut(
final VersionedStoreClient<T> versionedStoreClient,
final Optional<Sensor> expiredRecordSensor,
final Bytes key,
final byte[] value,
final long timestamp
Expand All @@ -466,7 +484,6 @@ private <T extends VersionedStoreSegment> void doPut(
// continue search in segments
status = maybePutToSegments(
versionedStoreClient,
expiredRecordSensor,
key,
value,
timestamp,
Expand All @@ -482,7 +499,6 @@ private <T extends VersionedStoreSegment> void doPut(
// or segments store). insert based on foundTs here instead.
finishPut(
versionedStoreClient,
expiredRecordSensor,
key,
value,
timestamp,
Expand Down Expand Up @@ -578,7 +594,6 @@ private <T extends VersionedStoreSegment> PutStatus maybePutToLatestValueStore(

private <T extends VersionedStoreSegment> PutStatus maybePutToSegments(
final VersionedStoreClient<T> versionedStoreClient,
final Optional<Sensor> expiredRecordSensor,
final Bytes key,
final byte[] value,
final long timestamp,
Expand Down Expand Up @@ -615,11 +630,10 @@ private <T extends VersionedStoreSegment> PutStatus maybePutToSegments(
}

if (foundMinTs < observedStreamTime - historyRetention) {
// the record being inserted does not affect version history. discard and return
if (expiredRecordSensor.isPresent()) {
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.

@mjsax mjsax Feb 14, 2023

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

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

Same below (2x).

@vcrfxia vcrfxia Feb 15, 2023

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

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.)

// this can happen during restore because individual put calls are executed after
// observedStreamTime is fast-forwarded for the entire batch of records being
// restored at once.
return new PutStatus(true, foundTs);
}

Expand Down Expand Up @@ -714,7 +728,6 @@ private <T extends VersionedStoreSegment> void putToSegment(

private <T extends VersionedStoreSegment> void finishPut(
final VersionedStoreClient<T> versionedStoreClient,
final Optional<Sensor> expiredRecordSensor,
final Bytes key,
final byte[] value,
final long timestamp,
Expand All @@ -730,10 +743,10 @@ private <T extends VersionedStoreSegment> void finishPut(
final T segment = versionedStoreClient.getOrCreateSegmentIfLive(
versionedStoreClient.segmentIdForTimestamp(timestamp), context, observedStreamTime);
if (segment == null) {
if (expiredRecordSensor.isPresent()) {
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.
// this can happen during restore because individual put calls are executed after
// observedStreamTime is fast-forwarded for the entire batch of records being
// restored at once.
return;
}

Expand Down Expand Up @@ -777,10 +790,10 @@ private <T extends VersionedStoreSegment> void finishPut(
final T segment = versionedStoreClient.getOrCreateSegmentIfLive(
versionedStoreClient.segmentIdForTimestamp(foundTs), context, observedStreamTime);
if (segment == null) {
if (expiredRecordSensor.isPresent()) {
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.
// this can happen during restore because individual put calls are executed after
// observedStreamTime is fast-forwarded for the entire batch of records being
// restored at once.
return;
}

Expand Down
71 changes: 71 additions & 0 deletions ...ams/src/test/java/org/apache/kafka/streams/state/internals/RocksDBVersionedStoreTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -47,6 +47,7 @@ public class RocksDBVersionedStoreTest {
private static final String STORE_NAME = "myversionedrocks";
private static final String METRICS_SCOPE = "versionedrocksdb";
private static final long HISTORY_RETENTION = 300_000L;
private static final long GRACE_PERIOD = HISTORY_RETENTION; // history retention doubles as grace period for now
private static final long SEGMENT_INTERVAL = HISTORY_RETENTION / 3;
private static final long BASE_TIMESTAMP = 10L;
private static final Serializer<String> STRING_SERIALIZER = new StringSerializer();
Expand Down Expand Up @@ -344,6 +345,19 @@ public void shouldDelete() {
assertThat(deleted.timestamp(), equalTo(SEGMENT_INTERVAL + 20));
}

@Test
public void shouldNotPutExpired() {
putToStore("k", "v", HISTORY_RETENTION + 10);

// grace period has not elapsed
putToStore("k1", "v1", HISTORY_RETENTION + 10 - GRACE_PERIOD);
verifyGetValueFromStore("k1", "v1", HISTORY_RETENTION + 10 - GRACE_PERIOD);

// grace period has elapsed, so this put does not take place
putToStore("k2", "v2", HISTORY_RETENTION + 9 - GRACE_PERIOD);
verifyGetNullFromStore("k2");
}

@Test
public void shouldGetFromOlderSegments() {
// use a different key to create three different segments
Expand Down Expand Up @@ -523,6 +537,63 @@ public void shouldRestoreMultipleBatches() {
verifyTimestampedGetNullFromStore("k", SEGMENT_INTERVAL - 15);
}

@Test
public void shouldNotRestoreExpired() {
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

@mjsax mjsax Feb 14, 2023

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

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

@vcrfxia vcrfxia Feb 15, 2023

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

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

@mjsax mjsax Feb 15, 2023

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

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.


store.restoreBatch(getChangelogRecords(records));

verifyGetValueFromStore("k", "v", HISTORY_RETENTION + 10);
verifyGetValueFromStore("k1", "v1", HISTORY_RETENTION + 10 - GRACE_PERIOD);
verifyGetNullFromStore("k2");
}

@Test
public void shouldRestoreEvenIfRecordWouldBeExpiredByEndOfBatch() {
final List<DataRecord> records = new ArrayList<>();
records.add(new DataRecord("k2", "v2", HISTORY_RETENTION - GRACE_PERIOD)); // this record will be older than grace period by the end of the batch, but should still be restored
records.add(new DataRecord("k", "v", HISTORY_RETENTION + 10));

store.restoreBatch(getChangelogRecords(records));

verifyGetValueFromStore("k2", "v2", HISTORY_RETENTION - GRACE_PERIOD);
verifyGetValueFromStore("k", "v", HISTORY_RETENTION + 10);
}

@Test
public void shouldAllowZeroHistoryRetention() {

@vcrfxia vcrfxia Feb 14, 2023

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

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.

// recreate store with zero history retention
store.close();
store = new RocksDBVersionedStore(STORE_NAME, METRICS_SCOPE, 0L, SEGMENT_INTERVAL);
store.init((StateStoreContext) context, store);

// put, get, and delete
putToStore("k", "v", BASE_TIMESTAMP);
verifyGetValueFromStore("k", "v", BASE_TIMESTAMP);
verifyTimestampedGetValueFromStore("k", BASE_TIMESTAMP, "v", BASE_TIMESTAMP);
verifyTimestampedGetValueFromStore("k", BASE_TIMESTAMP + 1, "v", BASE_TIMESTAMP); // query in "future" is allowed

// update existing record at same timestamp
putToStore("k", "updated", BASE_TIMESTAMP);
verifyGetValueFromStore("k", "updated", BASE_TIMESTAMP);
verifyTimestampedGetValueFromStore("k", BASE_TIMESTAMP, "updated", BASE_TIMESTAMP);

// put new record version
putToStore("k", "v2", BASE_TIMESTAMP + 2);
verifyGetValueFromStore("k", "v2", BASE_TIMESTAMP + 2);
verifyTimestampedGetValueFromStore("k", BASE_TIMESTAMP + 2, "v2", BASE_TIMESTAMP + 2);

// query in past (history retention expired) returns null
verifyTimestampedGetNullFromStore("k", BASE_TIMESTAMP + 1);

// put in past (grace period expired) does not update the store

@mjsax mjsax Feb 14, 2023

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

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

@vcrfxia vcrfxia Feb 15, 2023

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

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

@vcrfxia vcrfxia Feb 15, 2023

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

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?

@mjsax mjsax Feb 15, 2023

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

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

@mjsax mjsax Feb 15, 2023

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

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").

putToStore("k2", "v", BASE_TIMESTAMP + 1);
verifyGetNullFromStore("k2");
}

private void putToStore(final String key, final String value, final long timestamp) {
store.put(
new Bytes(STRING_SERIALIZER.serialize(null, key)),
Expand Down