vvcephei commented on a change in pull request #9293:
URL: https://github.com/apache/kafka/pull/9293#discussion_r489632127
##########
File path:
streams/src/main/java/org/apache/kafka/streams/state/internals/MeteredKeyValueStore.java
##########
@@ -46,43 +46,90 @@
* @param <K>
* @param <V>
*/
-public class MeteredKeyValueStore<K, V>
- extends WrappedStateStore<KeyValueStore<Bytes, byte[]>, K, V>
+public class MeteredKeyValueStore<K, V, OwnType extends
MeteredKeyValueStore<K, V, OwnType>>
Review comment:
The self-type reference is pretty mind-bending, but basically what it's
for is to let `reWrap` return the same type you call it on. So,
`MeteredKeyValueStore#reWrap` returns a `MeteredKeyValueStore` with the same
generic types, and `MeteredTimestampedKeyValueStore#reWrap` returns a
`MeteredTimestampedKeyValueStore` with the same generic parameters.
##########
File path:
streams/src/main/java/org/apache/kafka/streams/state/internals/MeteredKeyValueStore.java
##########
@@ -46,43 +46,90 @@
* @param <K>
* @param <V>
*/
-public class MeteredKeyValueStore<K, V>
- extends WrappedStateStore<KeyValueStore<Bytes, byte[]>, K, V>
+public class MeteredKeyValueStore<K, V, OwnType extends
MeteredKeyValueStore<K, V, OwnType>>
+ extends MeteredStore<K, V, MeteredKeyValueStore<K, V, OwnType>,
KeyValueStore<Bytes, byte[]>>
implements KeyValueStore<K, V> {
final Serde<K> keySerde;
final Serde<V> valueSerde;
StateSerdes<K, V> serdes;
- private final String metricsScope;
+ protected final String metricsScope;
protected final Time time;
protected Sensor putSensor;
- private Sensor putIfAbsentSensor;
+ protected Sensor putIfAbsentSensor;
protected Sensor getSensor;
- private Sensor deleteSensor;
- private Sensor putAllSensor;
- private Sensor allSensor;
- private Sensor rangeSensor;
- private Sensor flushSensor;
- private Sensor e2eLatencySensor;
- private ProcessorContext context;
- private StreamsMetricsImpl streamsMetrics;
- private final String threadId;
- private String taskId;
+ protected Sensor deleteSensor;
+ protected Sensor putAllSensor;
+ protected Sensor allSensor;
+ protected Sensor rangeSensor;
+ protected Sensor flushSensor;
+ protected Sensor e2eLatencySensor;
+ protected ProcessorContext context;
+ protected StreamsMetricsImpl streamsMetrics;
+ protected final String threadId;
+ protected String taskId;
MeteredKeyValueStore(final KeyValueStore<Bytes, byte[]> inner,
final String metricsScope,
final Time time,
final Serde<K> keySerde,
final Serde<V> valueSerde) {
+ this(
+ inner,
+ metricsScope,
+ Thread.currentThread().getName(),
+ time != null ? time : Time.SYSTEM,
+ keySerde,
+ valueSerde
+ );
+ }
+
+ protected MeteredKeyValueStore(final KeyValueStore<Bytes, byte[]> inner,
+ final String metricsScope,
+ final String threadId,
+ final Time time,
+ final Serde<K> keySerde,
+ final Serde<V> valueSerde) {
super(inner);
this.metricsScope = metricsScope;
- threadId = Thread.currentThread().getName();
- this.time = time != null ? time : Time.SYSTEM;
+ this.threadId = threadId;
+ this.time = time;
this.keySerde = keySerde;
this.valueSerde = valueSerde;
}
+ public MeteredKeyValueStore<K, V, OwnType> reWrap(final
KeyValueStore<Bytes, byte[]> inner) {
+ final MeteredKeyValueStore<K, V, OwnType> reWrapped = new
MeteredKeyValueStore<>(
+ inner,
+ metricsScope,
+ threadId,
+ time,
+ keySerde,
+ valueSerde
+ );
+
+ copyInit(reWrapped);
+
+ return reWrapped;
+ }
+
+ protected void copyInit(final MeteredKeyValueStore<K, V, OwnType>
reWrapped) {
Review comment:
We're basically cloning an already initialized Metered store, so we need
to copy over everything that gets set in init.
##########
File path:
streams/src/main/java/org/apache/kafka/streams/state/internals/StreamThreadStateStoreProvider.java
##########
@@ -84,26 +96,53 @@ public StreamThreadStateStoreProvider(final StreamThread
streamThread,
}
@SuppressWarnings("unchecked")
- private <T> T validateAndListStores(final StateStore store, final
QueryableStoreType<T> queryableStoreType, final String storeName, final TaskId
taskId) {
+ private static <T> T validateAndListStores(final StateStore store,
+ final QueryableStoreType<T>
queryableStoreType,
+ final String storeName,
+ final TaskId taskId,
+ final boolean bypassCache) {
if (store != null && queryableStoreType.accepts(store)) {
if (!store.isOpen()) {
throw new InvalidStateStoreException(
"Cannot get state store " + storeName + " for task " +
taskId +
" because the store is not open. " +
"The state store may have migrated to another
instances.");
}
+ final StateStore maybeUncached = maybeBypassCache(store,
bypassCache);
if (store instanceof TimestampedKeyValueStore &&
queryableStoreType instanceof QueryableStoreTypes.KeyValueStoreType) {
- return (T) new
ReadOnlyKeyValueStoreFacade<>((TimestampedKeyValueStore<Object, Object>) store);
+ return (T) new
ReadOnlyKeyValueStoreFacade<>((TimestampedKeyValueStore<Object, Object>)
maybeUncached);
} else if (store instanceof TimestampedWindowStore &&
queryableStoreType instanceof QueryableStoreTypes.WindowStoreType) {
- return (T) new
ReadOnlyWindowStoreFacade<>((TimestampedWindowStore<Object, Object>) store);
+ return (T) new
ReadOnlyWindowStoreFacade<>((TimestampedWindowStore<Object, Object>)
maybeUncached);
} else {
- return (T) store;
+ return (T) maybeUncached;
}
} else {
return null;
}
}
+ @SuppressWarnings({"rawtypes", "unchecked"})
+ private static StateStore maybeBypassCache(final StateStore store, final
boolean bypassCache) {
Review comment:
Here's where we actually bypass the cache. I've just realized that
`reWrap` should be in `WrappedStateStore`, in which case, this logic should
change a little, but the basic idea is there.
##########
File path:
streams/src/main/java/org/apache/kafka/streams/state/internals/StreamThreadStateStoreProvider.java
##########
@@ -84,26 +96,53 @@ public StreamThreadStateStoreProvider(final StreamThread
streamThread,
}
@SuppressWarnings("unchecked")
- private <T> T validateAndListStores(final StateStore store, final
QueryableStoreType<T> queryableStoreType, final String storeName, final TaskId
taskId) {
+ private static <T> T validateAndListStores(final StateStore store,
+ final QueryableStoreType<T>
queryableStoreType,
+ final String storeName,
+ final TaskId taskId,
+ final boolean bypassCache) {
if (store != null && queryableStoreType.accepts(store)) {
if (!store.isOpen()) {
throw new InvalidStateStoreException(
"Cannot get state store " + storeName + " for task " +
taskId +
" because the store is not open. " +
"The state store may have migrated to another
instances.");
}
+ final StateStore maybeUncached = maybeBypassCache(store,
bypassCache);
if (store instanceof TimestampedKeyValueStore &&
queryableStoreType instanceof QueryableStoreTypes.KeyValueStoreType) {
- return (T) new
ReadOnlyKeyValueStoreFacade<>((TimestampedKeyValueStore<Object, Object>) store);
+ return (T) new
ReadOnlyKeyValueStoreFacade<>((TimestampedKeyValueStore<Object, Object>)
maybeUncached);
} else if (store instanceof TimestampedWindowStore &&
queryableStoreType instanceof QueryableStoreTypes.WindowStoreType) {
- return (T) new
ReadOnlyWindowStoreFacade<>((TimestampedWindowStore<Object, Object>) store);
+ return (T) new
ReadOnlyWindowStoreFacade<>((TimestampedWindowStore<Object, Object>)
maybeUncached);
} else {
- return (T) store;
+ return (T) maybeUncached;
}
} else {
return null;
}
}
+ @SuppressWarnings({"rawtypes", "unchecked"})
+ private static StateStore maybeBypassCache(final StateStore store, final
boolean bypassCache) {
+ if (!bypassCache) {
+ return store;
+ } else if (store instanceof MeteredStore) {
+ final MeteredStore meteredStore = (MeteredStore) store;
+ final StateStore wrapped = meteredStore.wrapped();
+ final StateStore uncached = maybeBypassCache(wrapped, bypassCache);
+ if (uncached == wrapped) {
+ return store;
+ } else {
+ return meteredStore.reWrap(uncached);
Review comment:
And this is where, if we did actually strip out the caching layer, we
restore the layers "above" it.
##########
File path:
streams/src/main/java/org/apache/kafka/streams/state/internals/MeteredKeyValueStore.java
##########
@@ -46,43 +46,90 @@
* @param <K>
* @param <V>
*/
-public class MeteredKeyValueStore<K, V>
- extends WrappedStateStore<KeyValueStore<Bytes, byte[]>, K, V>
+public class MeteredKeyValueStore<K, V, OwnType extends
MeteredKeyValueStore<K, V, OwnType>>
+ extends MeteredStore<K, V, MeteredKeyValueStore<K, V, OwnType>,
KeyValueStore<Bytes, byte[]>>
implements KeyValueStore<K, V> {
final Serde<K> keySerde;
final Serde<V> valueSerde;
StateSerdes<K, V> serdes;
- private final String metricsScope;
+ protected final String metricsScope;
Review comment:
need to be protected so that we can set them in `copyInit`
##########
File path:
streams/src/main/java/org/apache/kafka/streams/state/internals/MeteredTimestampedKeyValueStore.java
##########
@@ -50,6 +50,31 @@
super(inner, metricScope, time, keySerde, valueSerde);
}
+ private MeteredTimestampedKeyValueStore(final KeyValueStore<Bytes, byte[]>
inner,
+ final String metricsScope,
+ final String threadId,
+ final Time time,
+ final Serde<K> keySerde,
+ final Serde<ValueAndTimestamp<V>>
valueSerde) {
+ super(inner, metricsScope, threadId, time, keySerde, valueSerde);
+ }
+
+ @Override
+ public MeteredTimestampedKeyValueStore<K, V> reWrap(final
KeyValueStore<Bytes, byte[]> inner) {
Review comment:
Overriding this here lets us preserve the correct type of the store when
re-wrapping.
##########
File path:
streams/src/main/java/org/apache/kafka/streams/state/internals/MeteredStore.java
##########
@@ -0,0 +1,40 @@
+/*
+ * 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.state.internals;
+
+import org.apache.kafka.common.serialization.Serde;
+import org.apache.kafka.streams.processor.StateStore;
+import org.apache.kafka.streams.state.KeyValueStore;
+
+/**
+ * A Metered {@link KeyValueStore} wrapper that is used for recording
operation metrics, and hence its
+ * inner KeyValueStore implementation do not need to provide its own metrics
collecting functionality.
+ * The inner {@link KeyValueStore} of this class is of type
<Bytes,byte[]>, hence we use {@link Serde}s
+ * to convert from <K,V> to <Bytes,byte[]>
+ *
+ * @param <K>
+ * @param <V>
+ */
+public abstract class MeteredStore<K, V, OwnType extends MeteredStore<K, V,
OwnType, InnerType>, InnerType extends StateStore>
+ extends WrappedStateStore<InnerType, K, V> {
+
+ public MeteredStore(final InnerType wrapped) {
+ super(wrapped);
+ }
+
+ abstract OwnType reWrap(final InnerType inner);
Review comment:
In retrospect, this actually belongs in `WrappedStateStore`. I'd fix it
if we choose to go this way.
##########
File path:
streams/src/main/java/org/apache/kafka/streams/state/internals/MeteredKeyValueStore.java
##########
@@ -46,43 +46,90 @@
* @param <K>
* @param <V>
*/
-public class MeteredKeyValueStore<K, V>
- extends WrappedStateStore<KeyValueStore<Bytes, byte[]>, K, V>
+public class MeteredKeyValueStore<K, V, OwnType extends
MeteredKeyValueStore<K, V, OwnType>>
+ extends MeteredStore<K, V, MeteredKeyValueStore<K, V, OwnType>,
KeyValueStore<Bytes, byte[]>>
implements KeyValueStore<K, V> {
final Serde<K> keySerde;
final Serde<V> valueSerde;
StateSerdes<K, V> serdes;
- private final String metricsScope;
+ protected final String metricsScope;
protected final Time time;
protected Sensor putSensor;
- private Sensor putIfAbsentSensor;
+ protected Sensor putIfAbsentSensor;
protected Sensor getSensor;
- private Sensor deleteSensor;
- private Sensor putAllSensor;
- private Sensor allSensor;
- private Sensor rangeSensor;
- private Sensor flushSensor;
- private Sensor e2eLatencySensor;
- private ProcessorContext context;
- private StreamsMetricsImpl streamsMetrics;
- private final String threadId;
- private String taskId;
+ protected Sensor deleteSensor;
+ protected Sensor putAllSensor;
+ protected Sensor allSensor;
+ protected Sensor rangeSensor;
+ protected Sensor flushSensor;
+ protected Sensor e2eLatencySensor;
+ protected ProcessorContext context;
+ protected StreamsMetricsImpl streamsMetrics;
+ protected final String threadId;
+ protected String taskId;
MeteredKeyValueStore(final KeyValueStore<Bytes, byte[]> inner,
final String metricsScope,
final Time time,
final Serde<K> keySerde,
final Serde<V> valueSerde) {
+ this(
+ inner,
+ metricsScope,
+ Thread.currentThread().getName(),
+ time != null ? time : Time.SYSTEM,
+ keySerde,
+ valueSerde
+ );
+ }
+
+ protected MeteredKeyValueStore(final KeyValueStore<Bytes, byte[]> inner,
+ final String metricsScope,
+ final String threadId,
+ final Time time,
+ final Serde<K> keySerde,
+ final Serde<V> valueSerde) {
super(inner);
this.metricsScope = metricsScope;
- threadId = Thread.currentThread().getName();
- this.time = time != null ? time : Time.SYSTEM;
+ this.threadId = threadId;
+ this.time = time;
this.keySerde = keySerde;
this.valueSerde = valueSerde;
}
+ public MeteredKeyValueStore<K, V, OwnType> reWrap(final
KeyValueStore<Bytes, byte[]> inner) {
Review comment:
This is the basic way that the cache bypass is implemented. You descend
the stack of wrapped stores until you find the cache, then you pop it off, then
you re-wrap the remaining store layers.
##########
File path:
streams/src/main/java/org/apache/kafka/streams/state/internals/StreamThreadStateStoreProvider.java
##########
@@ -84,26 +96,53 @@ public StreamThreadStateStoreProvider(final StreamThread
streamThread,
}
@SuppressWarnings("unchecked")
- private <T> T validateAndListStores(final StateStore store, final
QueryableStoreType<T> queryableStoreType, final String storeName, final TaskId
taskId) {
+ private static <T> T validateAndListStores(final StateStore store,
+ final QueryableStoreType<T>
queryableStoreType,
+ final String storeName,
+ final TaskId taskId,
+ final boolean bypassCache) {
if (store != null && queryableStoreType.accepts(store)) {
if (!store.isOpen()) {
throw new InvalidStateStoreException(
"Cannot get state store " + storeName + " for task " +
taskId +
" because the store is not open. " +
"The state store may have migrated to another
instances.");
}
+ final StateStore maybeUncached = maybeBypassCache(store,
bypassCache);
if (store instanceof TimestampedKeyValueStore &&
queryableStoreType instanceof QueryableStoreTypes.KeyValueStoreType) {
- return (T) new
ReadOnlyKeyValueStoreFacade<>((TimestampedKeyValueStore<Object, Object>) store);
+ return (T) new
ReadOnlyKeyValueStoreFacade<>((TimestampedKeyValueStore<Object, Object>)
maybeUncached);
} else if (store instanceof TimestampedWindowStore &&
queryableStoreType instanceof QueryableStoreTypes.WindowStoreType) {
- return (T) new
ReadOnlyWindowStoreFacade<>((TimestampedWindowStore<Object, Object>) store);
+ return (T) new
ReadOnlyWindowStoreFacade<>((TimestampedWindowStore<Object, Object>)
maybeUncached);
} else {
- return (T) store;
+ return (T) maybeUncached;
}
} else {
return null;
}
}
+ @SuppressWarnings({"rawtypes", "unchecked"})
+ private static StateStore maybeBypassCache(final StateStore store, final
boolean bypassCache) {
+ if (!bypassCache) {
+ return store;
+ } else if (store instanceof MeteredStore) {
+ final MeteredStore meteredStore = (MeteredStore) store;
+ final StateStore wrapped = meteredStore.wrapped();
+ final StateStore uncached = maybeBypassCache(wrapped, bypassCache);
+ if (uncached == wrapped) {
+ return store;
+ } else {
+ return meteredStore.reWrap(uncached);
+ }
+ } else if (WrappedStateStore.isCachingLayer(store)) {
+ return ((WrappedStateStore) store).wrapped();
Review comment:
Specifically, this is where we drop the caching layer from the returned
store hierarchy.
##########
File path:
streams/src/test/java/org/apache/kafka/streams/integration/QueryableStateIntegrationTest.java
##########
@@ -800,6 +811,76 @@ public void shouldBeAbleToQueryMapValuesAfterFilterState()
throws Exception {
}
}
+ @Test
+ public void shouldQueryKeyValueWithCacheBypassed() throws Exception {
Review comment:
This is the test I used to verify the behavior. Note: this test does not
verify that the cache is skipped, just that the query works with the cache
skipped. I set breakpoints in the code to additionally verify that the cache is
actually skipped.
If we go this route, I'd add unit tests so we can mechanically verify that
we're skipping 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.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
