[GitHub] [kafka] mjsax commented on a diff in pull request #13409: KAFKA-14491: [18/N] Update versioned store to check latest value on timestamped get

["mjsax (via GitHub)" <gi...@apache.org>]

mjsax commented on code in PR #13409:
URL: https://github.com/apache/kafka/pull/13409#discussion_r1139545717


##########
streams/src/main/java/org/apache/kafka/streams/state/VersionedKeyValueStore.java:
##########
@@ -59,12 +59,15 @@
      * Delete the value associated with this key from the store, at the specified timestamp
      * (if there is such a value), and return the deleted value.
      * <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.
+     * will not be performed and {@code null} will be returned.
+     * <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)}, with
+     * a caveat that if the deletion timestamp is older than the store's grace period
+     * (i.e., history retention) then the return value is always {@code null}, regardless
+     * of what {@link #get(Object, long) #get(key, timestamp)} would return.
      *

Review Comment:
   Add a new paragraph (as discussed in person):
   ```
   <p>
   Note: If you want to delete the latest version of a key, you should not do {@link #get(key)} and use the returned {@code validFrom} timestamp to call {@link #delete(key, timestamp) delete(key, validFrom)}. The same applies if you want to delete a specific version that was retrieved via {@link #get(key, timestamp)}. Instead, you should pass a business logic inferred timestamp hat specified when the delete actually happens. For example, it could be the timestamp of the currently processed input record or current stream-time.
   ```
   Something like this -- fee free to reword.



##########
streams/src/main/java/org/apache/kafka/streams/state/internals/RocksDBVersionedStore.java:
##########
@@ -156,9 +171,22 @@ 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 exceeded. check latest value store only

Review Comment:
   This might need more explanation:
   ```
   // if there is only a single version for a key which happens to be the latest version,
   // the data did not expire yet, and we need to return it
   ```



##########
streams/src/test/java/org/apache/kafka/streams/state/internals/RocksDBVersionedStoreTest.java:
##########
@@ -374,6 +374,24 @@ public void shouldNotPutExpired() {
         verifyExpiredRecordSensor(1);
     }
 
+    @Test
+    public void shouldNotDeleteExpired() {
+        putToStore("k1", "v1", 1L);
+        putToStore("k2", "v2", 1L);
+        putToStore("kother", "vother", HISTORY_RETENTION + 10); // use separate key to advance stream time
+
+        // grace period has not elapsed
+        VersionedRecord<String> deleted = deleteFromStore("k1", HISTORY_RETENTION + 10 - GRACE_PERIOD);
+        assertThat(deleted.value(), equalTo("v1"));
+        assertThat(deleted.timestamp(), equalTo(1L));
+
+        // grace period has elapsed, so this delete does not take place
+        deleted = deleteFromStore("k2", HISTORY_RETENTION + 9 - GRACE_PERIOD);
+        assertThat(deleted, nullValue()); // return value is null even though record exists because delete did not take place

Review Comment:
   I think we should verify that we can still `get(k2)` ?



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org