You are viewing a plain text version of this content. The canonical link for it is here.
Posted to jira@kafka.apache.org by GitBox <gi...@apache.org> on 2022/01/17 09:45:30 UTC
[GitHub] [kafka] vpapavas opened a new pull request #11682: feat: Add IQv2 query handling to the caching layer
vpapavas opened a new pull request #11682:
URL: https://github.com/apache/kafka/pull/11682
*More detailed description of your change,
if necessary. The PR title and PR message become
the squashed commit message, so use a separate
comment to ping reviewers.*
*Summary of testing strategy (including rationale)
for the feature or bug fix. Unit and/or integration
tests are expected for any behaviour change and
system tests should be considered for larger changes.*
### Committer Checklist (excluded from commit message)
- [ ] Verify design and implementation
- [ ] Verify test coverage and CI build status
- [ ] Verify documentation (including upgrade notes)
--
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
[GitHub] [kafka] vpapavas commented on a change in pull request #11682: KAFKA-13524: Add IQv2 query handling to the caching layer
Posted by GitBox <gi...@apache.org>.
vpapavas commented on a change in pull request #11682:
URL: https://github.com/apache/kafka/pull/11682#discussion_r787725141
##########
File path: streams/src/main/java/org/apache/kafka/streams/state/internals/CachingKeyValueStore.java
##########
@@ -83,10 +90,67 @@ public void init(final StateStoreContext context,
streamThread = Thread.currentThread();
}
- Position getPosition() {
+ public Position getPosition() {
return position;
Review comment:
I am not so sure about this. We won't have a way to know at what position the cache is, even just for debugging purposes.
--
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
[GitHub] [kafka] vpapavas commented on a change in pull request #11682: KAFKA-13524: Add IQv2 query handling to the caching layer
Posted by GitBox <gi...@apache.org>.
vpapavas commented on a change in pull request #11682:
URL: https://github.com/apache/kafka/pull/11682#discussion_r787752628
##########
File path: streams/src/main/java/org/apache/kafka/streams/processor/StateStore.java
##########
@@ -148,8 +150,12 @@ default void init(final StateStoreContext context, final StateStore root) {
default <R> QueryResult<R> query(
Query<R> query,
PositionBound positionBound,
- boolean collectExecutionInfo) {
+ QueryConfig config) {
// If a store doesn't implement a query handler, then all queries are unknown.
return QueryResult.forUnknownQueryType(query, this);
}
+
+ default Position getPosition() {
+ return Position.emptyPosition();
+ }
Review comment:
I thought I already did that? Can you give me an example of state store that I am missing? I thought that all state stores that need to handle the position were already covered by our previous PRs
--
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
[GitHub] [kafka] vpapavas commented on a change in pull request #11682: KAFKA-13524: Add IQv2 query handling to the caching layer
Posted by GitBox <gi...@apache.org>.
vpapavas commented on a change in pull request #11682:
URL: https://github.com/apache/kafka/pull/11682#discussion_r787893711
##########
File path: streams/src/main/java/org/apache/kafka/streams/processor/StateStore.java
##########
@@ -148,8 +150,12 @@ default void init(final StateStoreContext context, final StateStore root) {
default <R> QueryResult<R> query(
Query<R> query,
PositionBound positionBound,
- boolean collectExecutionInfo) {
+ QueryConfig config) {
// If a store doesn't implement a query handler, then all queries are unknown.
return QueryResult.forUnknownQueryType(query, this);
}
+
+ default Position getPosition() {
+ return Position.emptyPosition();
+ }
Review comment:
I think I addressed this, lmkwyt
--
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
[GitHub] [kafka] vpapavas commented on a change in pull request #11682: KAFKA-13524: Add IQv2 query handling to the caching layer
Posted by GitBox <gi...@apache.org>.
vpapavas commented on a change in pull request #11682:
URL: https://github.com/apache/kafka/pull/11682#discussion_r787752628
##########
File path: streams/src/main/java/org/apache/kafka/streams/processor/StateStore.java
##########
@@ -148,8 +150,12 @@ default void init(final StateStoreContext context, final StateStore root) {
default <R> QueryResult<R> query(
Query<R> query,
PositionBound positionBound,
- boolean collectExecutionInfo) {
+ QueryConfig config) {
// If a store doesn't implement a query handler, then all queries are unknown.
return QueryResult.forUnknownQueryType(query, this);
}
+
+ default Position getPosition() {
+ return Position.emptyPosition();
+ }
Review comment:
I though I already did that? Can you give me an example of state store that I am missing? I thought that all state stores that need to handle the position were already covered by our previous PRs
--
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
[GitHub] [kafka] vvcephei merged pull request #11682: KAFKA-13524: Add IQv2 query handling to the caching layer
Posted by GitBox <gi...@apache.org>.
vvcephei merged pull request #11682:
URL: https://github.com/apache/kafka/pull/11682
--
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
[GitHub] [kafka] vpapavas commented on a change in pull request #11682: feat: Add IQv2 query handling to the caching layer
Posted by GitBox <gi...@apache.org>.
vpapavas commented on a change in pull request #11682:
URL: https://github.com/apache/kafka/pull/11682#discussion_r786669972
##########
File path: streams/src/main/java/org/apache/kafka/streams/processor/StateStore.java
##########
@@ -152,4 +153,8 @@ default void init(final StateStoreContext context, final StateStore root) {
// If a store doesn't implement a query handler, then all queries are unknown.
return QueryResult.forUnknownQueryType(query, this);
}
+
Review comment:
This is a public API change. I will need to update the KIP. We need this so that we can get the position of the underlying store in the cache and merge it with that of the cache
--
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
[GitHub] [kafka] vvcephei commented on a change in pull request #11682: KAFKA-13524: Add IQv2 query handling to the caching layer
Posted by GitBox <gi...@apache.org>.
vvcephei commented on a change in pull request #11682:
URL: https://github.com/apache/kafka/pull/11682#discussion_r788231897
##########
File path: streams/src/main/java/org/apache/kafka/streams/state/internals/ChangeLoggingKeyValueBytesStore.java
##########
@@ -145,4 +145,9 @@ public void putAll(final List<KeyValue<Bytes, byte[]>> entries) {
void log(final Bytes key, final byte[] value) {
context.logChange(name(), key, value, context.timestamp(), position);
}
+
+ @Override
+ public Position getPosition() {
+ return position;
Review comment:
Similar concern here... It seems like we have to merge this layer's position with the wrapped store's position to compute a correct result for this method.
##########
File path: streams/src/main/java/org/apache/kafka/streams/query/KeyQuery.java
##########
@@ -40,7 +42,16 @@ private KeyQuery(final K key) {
* @param <V> The type of the value that will be retrieved
*/
public static <K, V> KeyQuery<K, V> withKey(final K key) {
- return new KeyQuery<>(key);
+ return new KeyQuery<>(key, false);
+ }
+
+ /**
+ * Specifies that the cache should be skipped during query evaluation. This means, that the query will always
+ * get forwarded to the underlying store.
+ */
+ @SuppressWarnings("unchecked")
+ public <K, V> KeyQuery<K, V> skipCache() {
+ return new KeyQuery<>((K) key, true);
Review comment:
```suggestion
public KeyQuery<K, V> skipCache() {
return new KeyQuery<>(key, true);
```
I think you won't need the cast or the suppression if you just keep the K and V bound to the same values as the current instance (`this`).
##########
File path: streams/src/main/java/org/apache/kafka/streams/state/internals/CachingKeyValueStore.java
##########
@@ -83,10 +90,67 @@ public void init(final StateStoreContext context,
streamThread = Thread.currentThread();
}
- Position getPosition() {
+ public Position getPosition() {
return position;
Review comment:
Fair enough. Maybe merging the underlying position into the cache's position here would be too much, but it does seem necessary to merge them together as the result of this method. Otherwise, the cache's position will "shadow" the underlying store's, and callers would have no way to learn the total position of the layered store.
--
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
[GitHub] [kafka] vvcephei commented on pull request #11682: KAFKA-13524: Add IQv2 query handling to the caching layer
Posted by GitBox <gi...@apache.org>.
vvcephei commented on pull request #11682:
URL: https://github.com/apache/kafka/pull/11682#issuecomment-1021720193
Oy, it looks like there's a compile error after that conflict resolution.
--
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
[GitHub] [kafka] vpapavas commented on a change in pull request #11682: feat: Add IQv2 query handling to the caching layer
Posted by GitBox <gi...@apache.org>.
vpapavas commented on a change in pull request #11682:
URL: https://github.com/apache/kafka/pull/11682#discussion_r786849830
##########
File path: streams/src/main/java/org/apache/kafka/streams/query/QueryConfig.java
##########
@@ -0,0 +1,35 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.kafka.streams.query;
+
+public class QueryConfig {
+ final boolean collectExecutionInfo;
+ final boolean skipCache;
+
+ public QueryConfig(final boolean collectExecutionInfo, final boolean skipCache) {
Review comment:
Instead of adding an extra boolean argument to the `query` method signature, I opted for this since this is future-forward in case we want to add more configurations
--
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
[GitHub] [kafka] vpapavas commented on a change in pull request #11682: feat: Add IQv2 query handling to the caching layer
Posted by GitBox <gi...@apache.org>.
vpapavas commented on a change in pull request #11682:
URL: https://github.com/apache/kafka/pull/11682#discussion_r786672285
##########
File path: streams/src/main/java/org/apache/kafka/streams/state/internals/TimestampedKeyValueStoreBuilder.java
##########
@@ -78,7 +78,7 @@ public TimestampedKeyValueStoreBuilder(final KeyValueBytesStoreSupplier storeSup
if (!enableCaching) {
return inner;
}
- return new CachingKeyValueStore(inner);
+ return new CachingKeyValueStore(inner, true);
Review comment:
We need to know what format of values the cache stores, whether they are timestamped or not
--
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
[GitHub] [kafka] vpapavas commented on a change in pull request #11682: KAFKA-13524: Add IQv2 query handling to the caching layer
Posted by GitBox <gi...@apache.org>.
vpapavas commented on a change in pull request #11682:
URL: https://github.com/apache/kafka/pull/11682#discussion_r787752628
##########
File path: streams/src/main/java/org/apache/kafka/streams/processor/StateStore.java
##########
@@ -148,8 +150,12 @@ default void init(final StateStoreContext context, final StateStore root) {
default <R> QueryResult<R> query(
Query<R> query,
PositionBound positionBound,
- boolean collectExecutionInfo) {
+ QueryConfig config) {
// If a store doesn't implement a query handler, then all queries are unknown.
return QueryResult.forUnknownQueryType(query, this);
}
+
+ default Position getPosition() {
+ return Position.emptyPosition();
+ }
Review comment:
I thought I already did that? Can you give me an example of state store that I am missing? I thought that all state stores that need to handle the position were already covered by our previous PRs
--
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
[GitHub] [kafka] vpapavas commented on a change in pull request #11682: KAFKA-13524: Add IQv2 query handling to the caching layer
Posted by GitBox <gi...@apache.org>.
vpapavas commented on a change in pull request #11682:
URL: https://github.com/apache/kafka/pull/11682#discussion_r791953273
##########
File path: streams/src/main/java/org/apache/kafka/streams/state/internals/ChangeLoggingKeyValueBytesStore.java
##########
@@ -145,4 +145,9 @@ public void putAll(final List<KeyValue<Bytes, byte[]>> entries) {
void log(final Bytes key, final byte[] value) {
context.logChange(name(), key, value, context.timestamp(), position);
}
+
+ @Override
+ public Position getPosition() {
+ return position;
Review comment:
The changelogging stores don't have a position anymore since I added the public `getPosition`. The merging now is necessary only in the caching layer
--
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
[GitHub] [kafka] vpapavas commented on a change in pull request #11682: feat: Add IQv2 query handling to the caching layer
Posted by GitBox <gi...@apache.org>.
vpapavas commented on a change in pull request #11682:
URL: https://github.com/apache/kafka/pull/11682#discussion_r786669972
##########
File path: streams/src/main/java/org/apache/kafka/streams/processor/StateStore.java
##########
@@ -152,4 +153,8 @@ default void init(final StateStoreContext context, final StateStore root) {
// If a store doesn't implement a query handler, then all queries are unknown.
return QueryResult.forUnknownQueryType(query, this);
}
+
Review comment:
This is a public API change. I will need to update the KIP
--
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
[GitHub] [kafka] vpapavas commented on a change in pull request #11682: feat: Add IQv2 query handling to the caching layer
Posted by GitBox <gi...@apache.org>.
vpapavas commented on a change in pull request #11682:
URL: https://github.com/apache/kafka/pull/11682#discussion_r786848574
##########
File path: streams/src/main/java/org/apache/kafka/streams/processor/StateStore.java
##########
@@ -148,8 +150,12 @@ default void init(final StateStoreContext context, final StateStore root) {
default <R> QueryResult<R> query(
Query<R> query,
PositionBound positionBound,
- boolean collectExecutionInfo) {
+ QueryConfig config) {
Review comment:
This is a public API change. I will need to update the KIP. We need this so that we can skip the cache for example for windowed queries in ksql
--
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
[GitHub] [kafka] patrickstuedi commented on a change in pull request #11682: feat: Add IQv2 query handling to the caching layer
Posted by GitBox <gi...@apache.org>.
patrickstuedi commented on a change in pull request #11682:
URL: https://github.com/apache/kafka/pull/11682#discussion_r786116191
##########
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:
Doesn't this mean that we might potentially serve queries from cache in cases where the cache store is not up-to-date?
--
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
[GitHub] [kafka] patrickstuedi commented on a change in pull request #11682: feat: Add IQv2 query handling to the caching layer
Posted by GitBox <gi...@apache.org>.
patrickstuedi commented on a change in pull request #11682:
URL: https://github.com/apache/kafka/pull/11682#discussion_r786686638
##########
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:
Makes sense
--
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
[GitHub] [kafka] vvcephei commented on a change in pull request #11682: KAFKA-13524: Add IQv2 query handling to the caching layer
Posted by GitBox <gi...@apache.org>.
vvcephei commented on a change in pull request #11682:
URL: https://github.com/apache/kafka/pull/11682#discussion_r792216685
##########
File path: streams/src/main/java/org/apache/kafka/streams/state/internals/InMemoryTimeOrderedKeyValueBuffer.java
##########
@@ -238,13 +243,22 @@ private void init(final StateStore root) {
updateBufferMetrics();
open = true;
partition = context.taskId().partition();
+ consistencyEnabled = StreamsConfig.InternalConfig.getBoolean(
+ context.appConfigs(),
+ IQ_CONSISTENCY_OFFSET_VECTOR_ENABLED,
+ false);
}
@Override
public boolean isOpen() {
return open;
}
+ @Override
+ public Position getPosition() {
+ return position;
Review comment:
This isn't harmful, but it will also never be used, since this store is not exposed to IQ. I'm not sure if we still want this implementation or if we'd rather just throw an `UnsupportedOperationException`. I think that's something you can do in a follow-up PR if you like.
##########
File path: streams/src/main/java/org/apache/kafka/streams/processor/StateStore.java
##########
@@ -152,4 +153,8 @@ default void init(final StateStoreContext context, final StateStore root) {
// If a store doesn't implement a query handler, then all queries are unknown.
return QueryResult.forUnknownQueryType(query, this);
}
+
Review comment:
Thanks! Since it's an added method to an interface people are meant to implement, we need to add a suitable default. I'm planning to take care of that in https://github.com/apache/kafka/pull/11676 (which also adds this method), so no worries.
--
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
[GitHub] [kafka] vpapavas commented on a change in pull request #11682: feat: Add IQv2 query handling to the caching layer
Posted by 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
[GitHub] [kafka] vvcephei commented on a change in pull request #11682: feat: Add IQv2 query handling to the caching layer
Posted by GitBox <gi...@apache.org>.
vvcephei commented on a change in pull request #11682:
URL: https://github.com/apache/kafka/pull/11682#discussion_r787180989
##########
File path: streams/src/main/java/org/apache/kafka/streams/processor/StateStore.java
##########
@@ -148,8 +150,12 @@ default void init(final StateStoreContext context, final StateStore root) {
default <R> QueryResult<R> query(
Query<R> query,
PositionBound positionBound,
- boolean collectExecutionInfo) {
+ QueryConfig config) {
// If a store doesn't implement a query handler, then all queries are unknown.
return QueryResult.forUnknownQueryType(query, this);
}
+
+ default Position getPosition() {
+ return Position.emptyPosition();
+ }
Review comment:
Thanks, @vpapavas ! I agree with adding this to the API.
##########
File path: streams/src/main/java/org/apache/kafka/streams/query/StateQueryRequest.java
##########
@@ -39,21 +39,24 @@
private final Query<R> query;
private final boolean executionInfoEnabled;
private final boolean requireActive;
+ private final boolean skipCache;
private StateQueryRequest(
final String storeName,
final PositionBound position,
final Optional<Set<Integer>> partitions,
final Query<R> query,
final boolean executionInfoEnabled,
- final boolean requireActive) {
+ final boolean requireActive,
+ final boolean skipCache) {
Review comment:
Ditto here. If we just add it to KeyQuery, you won't need any changes to this class.
##########
File path: streams/src/main/java/org/apache/kafka/streams/query/QueryConfig.java
##########
@@ -0,0 +1,35 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.kafka.streams.query;
+
+public class QueryConfig {
+ final boolean collectExecutionInfo;
+ final boolean skipCache;
+
+ public QueryConfig(final boolean collectExecutionInfo, final boolean skipCache) {
Review comment:
Thanks! As mentioned above, I think we should actually add `skipCache` to the `KeyQuery` instead.
##########
File path: streams/src/main/java/org/apache/kafka/streams/processor/StateStore.java
##########
@@ -148,8 +150,12 @@ default void init(final StateStoreContext context, final StateStore root) {
default <R> QueryResult<R> query(
Query<R> query,
PositionBound positionBound,
- boolean collectExecutionInfo) {
+ QueryConfig config) {
// If a store doesn't implement a query handler, then all queries are unknown.
return QueryResult.forUnknownQueryType(query, this);
}
+
+ default Position getPosition() {
+ return Position.emptyPosition();
+ }
Review comment:
Since it will be in the public API and therefore available from IQv1 StateStores as well as StateStores obtained in processors via ProcessorContext#getStateStore, I suppose we should override the method in the relevant wrapping stores so that the method will return the correct value for all callers, right?
It'd be fine to do that in a follow-on PR, but let's file a 3.2 blocker ticket so we don't forget.
##########
File path: streams/src/main/java/org/apache/kafka/streams/state/internals/CachingKeyValueStore.java
##########
@@ -83,10 +90,67 @@ 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 QueryConfig config) {
+ if (config.isSkipCache()) {
+ return wrapped().query(query, positionBound, config);
+ }
+ final long start = config.isCollectExecutionInfo() ? 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;
+ final Bytes key = keyQuery.getKey();
+
+ validateStoreOpen();
+ final Lock lock = this.lock.readLock();
+ lock.lock();
+ try {
+ validateStoreOpen();
+ if (context.cache() != null) {
+ final LRUCacheEntry lruCacheEntry = context.cache().get(cacheName, key);
+ if (lruCacheEntry != null) {
+ final byte[] rawValue;
+ if (timestampedSchema && !WrappedStateStore.isTimestamped(wrapped())) {
+ rawValue = ValueAndTimestampDeserializer.rawValue(lruCacheEntry.value());
+ } else {
+ rawValue = lruCacheEntry.value();
+ }
+ result = (QueryResult<R>) QueryResult.forResult(rawValue);
+ }
+ }
+ } finally {
+ lock.unlock();
+ }
+
+ if (result == null) {
+ result = wrapped().query(query, positionBound, config);
Review comment:
I think we want to do this query unbounded, since the provided bound may include a position that hasn't been flushed yet. We've already validated the position above. Come to think of it, though, shouldn't we check the position after we grab the lock?
##########
File path: streams/src/main/java/org/apache/kafka/streams/state/internals/CachingKeyValueStore.java
##########
@@ -83,10 +90,67 @@ 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 QueryConfig config) {
+ if (config.isSkipCache()) {
+ return wrapped().query(query, positionBound, config);
+ }
+ final long start = config.isCollectExecutionInfo() ? 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());
Review comment:
If you like the suggestion on L93, we could just call `getPosition()` here.
##########
File path: streams/src/main/java/org/apache/kafka/streams/processor/StateStore.java
##########
@@ -148,8 +150,12 @@ default void init(final StateStoreContext context, final StateStore root) {
default <R> QueryResult<R> query(
Query<R> query,
PositionBound positionBound,
- boolean collectExecutionInfo) {
+ QueryConfig config) {
Review comment:
Thanks, @vpapavas ! I think we're better off with adding `skipCache` to the KeyQuery directly (otherwise, we would imply it applies to all queries when it does not).
However, I was previously on the fence about whether to add `collectExecutionInfo` as a parameter or whether to add a `QueryConfig` class just like this one to future-proof the API. Even though I don't think we need `skipCache` in the QueryConfig right now, the fact that it was almost necessary so soon makes me think I made the wrong call before.
I'd be +1 on keeping QueryConfig in the API.
##########
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:
Good thinking, all. As usual, I think the situation becomes much clearer if we mentally rename "cache" to "write buffer", since that's what it actually is.
##########
File path: streams/src/main/java/org/apache/kafka/streams/state/internals/CachingKeyValueStore.java
##########
@@ -83,10 +90,67 @@ public void init(final StateStoreContext context,
streamThread = Thread.currentThread();
}
- Position getPosition() {
+ public Position getPosition() {
return position;
Review comment:
Should we merge in the wrapped store's position for the same reason you merge in the wrapped store's position below?
--
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
[GitHub] [kafka] vvcephei commented on pull request #11682: KAFKA-13524: Add IQv2 query handling to the caching layer
Posted by GitBox <gi...@apache.org>.
vvcephei commented on pull request #11682:
URL: https://github.com/apache/kafka/pull/11682#issuecomment-1021786440
Turned out that the new test I added on trunk was incompatible with the `StoreQueryUtils.handleBasicQueries`. I fixed it and went ahead and added that default impl to `StateStore.getPosition`
--
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
[GitHub] [kafka] vvcephei commented on pull request #11682: KAFKA-13524: Add IQv2 query handling to the caching layer
Posted by GitBox <gi...@apache.org>.
vvcephei commented on pull request #11682:
URL: https://github.com/apache/kafka/pull/11682#issuecomment-1022316812
Unrelated test failures:
```
Build / ARM / kafka.server.ReplicaManagerTest.testDeltaFromLeaderToFollower()
Build / JDK 8 and Scala 2.12 / kafka.admin.LeaderElectionCommandTest.[1] Type=Raft, Name=testElectionResultOutput, Security=PLAINTEXT
Build / JDK 8 and Scala 2.12 / kafka.api.PlaintextAdminIntegrationTest.testReplicaCanFetchFromLogStartOffsetAfterDeleteRecords()
Build / JDK 8 and Scala 2.12 / kafka.api.TransactionsTest.testAbortTransactionTimeout()
Build / JDK 17 and Scala 2.13 / integration.kafka.server.FetchRequestBetweenDifferentIbpTest.testControllerOldToNewIBP()
Build / JDK 17 and Scala 2.13 / kafka.admin.LeaderElectionCommandTest.[1] Type=Raft, Name=testTopicPartition, Security=PLAINTEXT
```
--
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
[GitHub] [kafka] vvcephei commented on pull request #11682: KAFKA-13524: Add IQv2 query handling to the caching layer
Posted by GitBox <gi...@apache.org>.
vvcephei commented on pull request #11682:
URL: https://github.com/apache/kafka/pull/11682#issuecomment-1021685037
Huh, I just used the GH UI to fix some merge conflicts, and it looks like it chose to merge trunk into this branch :/
--
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