[GitHub] [kafka] vpapavas commented on a change in pull request #11682: feat: Add IQv2 query handling to the caching layer

[GitBox <gi...@apache.org>]

vpapavas commented on a change in pull request #11682:
URL: https://github.com/apache/kafka/pull/11682#discussion_r786669591



##########
File path: streams/src/main/java/org/apache/kafka/streams/state/internals/CachingKeyValueStore.java
##########
@@ -83,10 +89,64 @@ public void init(final StateStoreContext context,
         streamThread = Thread.currentThread();
     }
 
-    Position getPosition() {
+    public Position getPosition() {
         return position;
     }
 
+
+    @SuppressWarnings("unchecked")
+    @Override
+    public <R> QueryResult<R> query(final Query<R> query,
+                                    final PositionBound positionBound,
+                                    final boolean collectExecutionInfo) {
+        final long start = collectExecutionInfo ? System.nanoTime() : -1L;
+        QueryResult<R> result = null;
+        final int partition = context.taskId().partition();
+        final Position mergedPosition = Position.emptyPosition();
+        mergedPosition.merge(position);
+        mergedPosition.merge(wrapped().getPosition());
+        if (!StoreQueryUtils.isPermitted(mergedPosition, positionBound, partition)) {
+            result = QueryResult.notUpToBound(mergedPosition, positionBound, partition);
+        } else if (query instanceof KeyQuery) {
+            final KeyQuery<Bytes, byte[]> keyQuery = (KeyQuery<Bytes, byte[]>) query;

Review comment:
       The worry here is that the position of the store might be ahead of the cache, but the query will get answered from the cache. I see two scenarios where the position of the cache and the store can differ:
   1) The cache is ahead of the store which occurs during normal operation before the cache is flushed. In this case queries for keys that are in the cache will get answered by the cache, whereas keys that are not will be answered by the store. Getting the merged position handles this case.
   2) The cache is behind the store. This happens during restoring in which case the cache will be empty and all queries will be served by the store. Again, getting the merged position handles this case.




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