diff --git a/streams/src/main/java/org/apache/kafka/streams/state/VersionedKeyValueStore.java b/streams/src/main/java/org/apache/kafka/streams/state/VersionedKeyValueStore.java index 1cdf4ad8f31fe..a80dabcb0ba6e 100644 --- a/streams/src/main/java/org/apache/kafka/streams/state/VersionedKeyValueStore.java +++ b/streams/src/main/java/org/apache/kafka/streams/state/VersionedKeyValueStore.java @@ -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.) + *

+ * 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 The key type * @param The value type @@ -38,6 +43,10 @@ public interface VersionedKeyValueStore extends StateStore { /** * Add a new record version associated with the specified key and timestamp. + *

+ * 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. @@ -52,6 +61,10 @@ public interface VersionedKeyValueStore extends StateStore { *

* This operation is semantically equivalent to {@link #get(Object, long) #get(key, timestamp)} * followed by {@link #put(Object, Object, long) #put(key, null, timestamp)}. + *

+ * 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 diff --git a/streams/src/main/java/org/apache/kafka/streams/state/internals/RocksDBVersionedStore.java b/streams/src/main/java/org/apache/kafka/streams/state/internals/RocksDBVersionedStore.java index f56ceccc0b740..133dd9a10a4f8 100644 --- a/streams/src/main/java/org/apache/kafka/streams/state/internals/RocksDBVersionedStore.java +++ b/streams/src/main/java/org/apache/kafka/streams/state/internals/RocksDBVersionedStore.java @@ -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; @@ -82,6 +81,7 @@ public class RocksDBVersionedStore implements VersionedKeyValueStore= 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); @@ -111,15 +115,20 @@ public class RocksDBVersionedStore implements VersionedKeyValueStore get(final Bytes key) { public VersionedRecord 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; @@ -283,9 +293,13 @@ public void init(final StateStoreContext context, final StateStore root) { // VisibleForTesting void restoreBatch(final Collection> 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 record : records) { - observedStreamTime = Math.max(observedStreamTime, record.timestamp()); + endOfBatchStreamTime = Math.max(endOfBatchStreamTime, record.timestamp()); } final VersionedStoreClient restoreClient = restoreWriteBuffer.getClient(); @@ -297,6 +311,14 @@ void restoreBatch(final Collection> 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 record : records) { + if (record.timestamp() < observedStreamTime - gracePeriod) { + // record is older than grace period and was therefore never written to the store + 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, @@ -306,7 +328,7 @@ void restoreBatch(final Collection> records) { // put records to write buffer doPut( restoreClient, - Optional.empty(), + endOfBatchStreamTime, new Bytes(record.key()), record.value(), record.timestamp() @@ -438,9 +460,22 @@ public void writeLatestValues(final WriteBatch batch) throws RocksDBException { } } + /** + * Helper method shared between put and restore. + *

+ * 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. + */ private void doPut( final VersionedStoreClient versionedStoreClient, - final Optional expiredRecordSensor, + final long observedStreamTime, final Bytes key, final byte[] value, final long timestamp @@ -453,6 +488,7 @@ private void doPut( // check latest value store PutStatus status = maybePutToLatestValueStore( versionedStoreClient, + observedStreamTime, key, value, timestamp @@ -466,7 +502,7 @@ private void doPut( // continue search in segments status = maybePutToSegments( versionedStoreClient, - expiredRecordSensor, + observedStreamTime, key, value, timestamp, @@ -482,7 +518,7 @@ private void doPut( // or segments store). insert based on foundTs here instead. finishPut( versionedStoreClient, - expiredRecordSensor, + observedStreamTime, key, value, timestamp, @@ -515,6 +551,7 @@ private static class PutStatus { private PutStatus maybePutToLatestValueStore( final VersionedStoreClient versionedStoreClient, + final long observedStreamTime, final Bytes key, final byte[] value, final long timestamp @@ -578,7 +615,7 @@ private PutStatus maybePutToLatestValueStore( private PutStatus maybePutToSegments( final VersionedStoreClient versionedStoreClient, - final Optional expiredRecordSensor, + final long observedStreamTime, final Bytes key, final byte[] value, final long timestamp, @@ -605,6 +642,7 @@ private PutStatus maybePutToSegments( // insert and conclude the procedure. putToSegment( versionedStoreClient, + observedStreamTime, segment, rawSegmentValue, key, @@ -615,11 +653,10 @@ private 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. + // 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); } @@ -634,6 +671,7 @@ private PutStatus maybePutToSegments( private void putToSegment( final VersionedStoreClient versionedStoreClient, + final long observedStreamTime, final T segment, final byte[] rawSegmentValue, final Bytes key, @@ -714,7 +752,7 @@ private void putToSegment( private void finishPut( final VersionedStoreClient versionedStoreClient, - final Optional expiredRecordSensor, + final long observedStreamTime, final Bytes key, final byte[] value, final long timestamp, @@ -730,10 +768,10 @@ private 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; } @@ -777,10 +815,10 @@ private 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; } diff --git a/streams/src/test/java/org/apache/kafka/streams/state/internals/RocksDBVersionedStoreTest.java b/streams/src/test/java/org/apache/kafka/streams/state/internals/RocksDBVersionedStoreTest.java index fb97ad3a6b784..15d03ee94ef47 100644 --- a/streams/src/test/java/org/apache/kafka/streams/state/internals/RocksDBVersionedStoreTest.java +++ b/streams/src/test/java/org/apache/kafka/streams/state/internals/RocksDBVersionedStoreTest.java @@ -16,14 +16,20 @@ */ package org.apache.kafka.streams.state.internals; +import static org.apache.kafka.common.utils.Utils.mkEntry; +import static org.apache.kafka.common.utils.Utils.mkMap; import static org.hamcrest.CoreMatchers.equalTo; import static org.hamcrest.CoreMatchers.nullValue; import static org.hamcrest.MatcherAssert.assertThat; +import static org.junit.Assert.assertEquals; import java.util.ArrayList; import java.util.List; +import java.util.Map; import java.util.Optional; import org.apache.kafka.clients.consumer.ConsumerRecord; +import org.apache.kafka.common.Metric; +import org.apache.kafka.common.MetricName; import org.apache.kafka.common.header.internals.RecordHeaders; import org.apache.kafka.common.record.TimestampType; import org.apache.kafka.common.serialization.Deserializer; @@ -47,12 +53,16 @@ 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_SERIALIZER = new StringSerializer(); private static final Deserializer STRING_DESERIALIZER = new StringDeserializer(); + private static final String DROPPED_RECORDS_METRIC = "dropped-records-total"; + private static final String TASK_LEVEL_GROUP = "stream-task-metrics"; private InternalMockProcessorContext context; + private Map expectedMetricsTags; private RocksDBVersionedStore store; @@ -66,6 +76,11 @@ public void before() { ); context.setTime(BASE_TIMESTAMP); + expectedMetricsTags = mkMap( + mkEntry("thread-id", Thread.currentThread().getName()), + mkEntry("task-id", context.taskId().toString()) + ); + store = new RocksDBVersionedStore(STORE_NAME, METRICS_SCOPE, HISTORY_RETENTION, SEGMENT_INTERVAL); store.init((StateStoreContext) context, store); } @@ -344,6 +359,21 @@ 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"); + + verifyExpiredRecordSensor(1); + } + @Test public void shouldGetFromOlderSegments() { // use a different key to create three different segments @@ -523,6 +553,70 @@ public void shouldRestoreMultipleBatches() { verifyTimestampedGetNullFromStore("k", SEGMENT_INTERVAL - 15); } + @Test + public void shouldNotRestoreExpired() { + final List 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 + + store.restoreBatch(getChangelogRecords(records)); + + verifyGetValueFromStore("k", "v", HISTORY_RETENTION + 10); + verifyGetValueFromStore("k1", "v1", HISTORY_RETENTION + 10 - GRACE_PERIOD); + verifyGetNullFromStore("k2"); + + verifyExpiredRecordSensor(0); + } + + @Test + public void shouldRestoreEvenIfRecordWouldBeExpiredByEndOfBatch() { + final List 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() { + // recreate store with zero history retention + store.close(); + store = new RocksDBVersionedStore(STORE_NAME, METRICS_SCOPE, 0L, SEGMENT_INTERVAL); + store.init((StateStoreContext) context, store); + + // put and get + 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); + + // delete existing key + deleteFromStore("k", BASE_TIMESTAMP + 3); + verifyGetNullFromStore("k"); + + // put in past (grace period expired) does not update the store + putToStore("k2", "v", BASE_TIMESTAMP + 2); + verifyGetNullFromStore("k2"); + verifyExpiredRecordSensor(1); + } + private void putToStore(final String key, final String value, final long timestamp) { store.put( new Bytes(STRING_SERIALIZER.serialize(null, key)), @@ -571,6 +665,13 @@ private void verifyTimestampedGetNullFromStore(final String key, final long time assertThat(record, nullValue()); } + private void verifyExpiredRecordSensor(final int expectedValue) { + final Metric metric = context.metrics().metrics().get( + new MetricName(DROPPED_RECORDS_METRIC, TASK_LEVEL_GROUP, "", expectedMetricsTags) + ); + assertEquals((Double) metric.metricValue(), expectedValue, 0.001); + } + private static VersionedRecord deserializedRecord(final VersionedRecord versionedRecord) { return versionedRecord == null ? null