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 all commits
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
88 changes: 63 additions & 25 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,15 +115,20 @@ 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;
}
observedStreamTime = Math.max(observedStreamTime, timestamp);

doPut(
versionedStoreClient,
Optional.of(expiredRecordSensor),
observedStreamTime,
key,
value,
timestamp
);

observedStreamTime = Math.max(observedStreamTime, timestamp);
}

@Override
Expand Down Expand Up @@ -147,6 +156,7 @@ public VersionedRecord<byte[]> get(final Bytes key) {
public VersionedRecord<byte[]> get(final Bytes key, final long asOfTimestamp) {

if (asOfTimestamp < observedStreamTime - historyRetention) {
LOG.warn("Returning null for expired get.");
// history retention has elapsed. return null for predictability, even if data
// is still present in store.
return null;
Expand Down Expand Up @@ -283,9 +293,13 @@ 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

// compute the observed stream time at the end of the restore 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.
long endOfBatchStreamTime = observedStreamTime;
for (final ConsumerRecord<byte[], byte[]> record : records) {
observedStreamTime = Math.max(observedStreamTime, record.timestamp());
endOfBatchStreamTime = Math.max(endOfBatchStreamTime, record.timestamp());
}

final VersionedStoreClient<?> restoreClient = restoreWriteBuffer.getClient();
Expand All @@ -297,6 +311,14 @@ 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() < observedStreamTime - 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;
}
// advance observed stream time as usual, for use in deciding whether records have
// exceeded the store's grace period and should be dropped.
observedStreamTime = Math.max(observedStreamTime, record.timestamp());

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

/**
* Helper method shared between put and restore.
* <p>
* This method does not check whether the record being put is expired based on grace period
* or not; that is the caller's responsibility. This method does, however, check whether the
* record is expired based on history retention, by using the current
* {@code observedStreamTime}, and returns without inserting into the store if so. It can be
* possible that a record is not expired based on grace period but is expired based on
* 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.

@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 for adding this! Great addition!

*/
private <T extends VersionedStoreSegment> void doPut(
final VersionedStoreClient<T> versionedStoreClient,
final Optional<Sensor> expiredRecordSensor,
final long observedStreamTime,
final Bytes key,
final byte[] value,
final long timestamp
Expand All @@ -453,6 +488,7 @@ private <T extends VersionedStoreSegment> void doPut(
// check latest value store
PutStatus status = maybePutToLatestValueStore(
versionedStoreClient,
observedStreamTime,
key,
value,
timestamp
Expand All @@ -466,7 +502,7 @@ private <T extends VersionedStoreSegment> void doPut(
// continue search in segments
status = maybePutToSegments(
versionedStoreClient,
expiredRecordSensor,
observedStreamTime,
key,
value,
timestamp,
Expand All @@ -482,7 +518,7 @@ private <T extends VersionedStoreSegment> void doPut(
// or segments store). insert based on foundTs here instead.
finishPut(
versionedStoreClient,
expiredRecordSensor,
observedStreamTime,
key,
value,
timestamp,
Expand Down Expand Up @@ -515,6 +551,7 @@ private static class PutStatus {

private <T extends VersionedStoreSegment> PutStatus maybePutToLatestValueStore(
final VersionedStoreClient<T> versionedStoreClient,
final long observedStreamTime,
final Bytes key,
final byte[] value,
final long timestamp
Expand Down Expand Up @@ -578,7 +615,7 @@ private <T extends VersionedStoreSegment> PutStatus maybePutToLatestValueStore(

private <T extends VersionedStoreSegment> PutStatus maybePutToSegments(
final VersionedStoreClient<T> versionedStoreClient,
final Optional<Sensor> expiredRecordSensor,
final long observedStreamTime,
final Bytes key,
final byte[] value,
final long timestamp,
Expand All @@ -605,6 +642,7 @@ private <T extends VersionedStoreSegment> PutStatus maybePutToSegments(
// insert and conclude the procedure.
putToSegment(
versionedStoreClient,
observedStreamTime,
segment,
rawSegmentValue,
key,
Expand All @@ -615,11 +653,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 All @@ -634,6 +671,7 @@ private <T extends VersionedStoreSegment> PutStatus maybePutToSegments(

private <T extends VersionedStoreSegment> void putToSegment(
final VersionedStoreClient<T> versionedStoreClient,
final long observedStreamTime,
final T segment,
final byte[] rawSegmentValue,
final Bytes key,
Expand Down Expand Up @@ -714,7 +752,7 @@ private <T extends VersionedStoreSegment> void putToSegment(

private <T extends VersionedStoreSegment> void finishPut(
final VersionedStoreClient<T> versionedStoreClient,
final Optional<Sensor> expiredRecordSensor,
final long observedStreamTime,
final Bytes key,
final byte[] value,
final long timestamp,
Expand All @@ -730,10 +768,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 +815,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
Loading