patrickstuedi commented on a change in pull request #11567:
URL: https://github.com/apache/kafka/pull/11567#discussion_r773792800
##########
File path:
streams/src/main/java/org/apache/kafka/streams/state/internals/InMemorySessionStore.java
##########
@@ -317,14 +318,31 @@ public boolean isOpen() {
@Override
public <R> QueryResult<R> query(final Query<R> query, final PositionBound
positionBound,
final boolean collectExecutionInfo) {
+
+ if (query instanceof WindowRangeQuery) {
+ @SuppressWarnings("unchecked") final WindowRangeQuery<Bytes,
byte[]> windowRangeQuery = (WindowRangeQuery<Bytes, byte[]>) query;
+ if (windowRangeQuery.getKey().isPresent()) {
+ final Bytes key = windowRangeQuery.getKey().get();
+ final KeyValueIterator<Windowed<Bytes>, byte[]>
keyValueIterator = this.fetch(key);
+ @SuppressWarnings("unchecked") final R result = (R)
keyValueIterator;
+ final QueryResult<R> queryResult =
QueryResult.forResult(result);
+ return queryResult;
+ }
+ }
+
+ return null;
+
+ /*
return StoreQueryUtils.handleBasicQueries(
- query,
- positionBound,
- collectExecutionInfo,
- this,
- position,
- context.taskId().partition()
+ query,
+ positionBound,
+ collectExecutionInfo,
+ this,
+ position,
+ context.taskId().partition()
);
+
+ */
Review comment:
I was looking at your merged PR and coded the store query routines in a
similar spirit. For instance, I follow your practice in the MeteredKeyValue
store and applied it to MeteredWindow/Session stores. Similarly for the
InMemoryKeyValue store I noticed you handle range queries in place, so I did
the same for the InMemoryWindowStore. Generally if you feel we should handle
queries in the util class (as some are) then probably we should be consistent
across the stores and remove the local handlers and instead delegate to the
store utils?
--
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-unsubscr...@kafka.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
