aliehsaeedii commented on code in PR #22666:
URL: https://github.com/apache/kafka/pull/22666#discussion_r3475557031
##########
streams/src/main/java/org/apache/kafka/streams/state/internals/MeteredKeyValueStore.java:
##########
@@ -323,8 +323,11 @@ private <R> QueryResult<R> runKeyQuery(final Query<R>
query,
final QueryConfig config) {
final QueryResult<R> result;
final KeyQuery<K, V> typedKeyQuery = (KeyQuery<K, V>) query;
- final KeyQuery<Bytes, byte[]> rawKeyQuery =
+ KeyQuery<Bytes, byte[]> rawKeyQuery =
KeyQuery.withKey(serializeKey(typedKeyQuery.getKey()));
+ if (typedKeyQuery.isSkipCache()) {
Review Comment:
This propagates `skipCache` for every plain key-value store, not just
header-aware ones — `skipCache()` was silently a no-op on `KeyQuery` here
before, so this changes read-your-writes behavior PR-wide (same applies to the
equivalent change in `MeteredTimestampedKeyValueStore`). It looks like a
genuine fix, but it reaches beyond the "headers-aware stores" scope of
KIP-1356.
##########
streams/src/main/java/org/apache/kafka/streams/state/internals/MeteredTimestampedKeyValueStoreWithHeaders.java:
##########
@@ -381,6 +390,45 @@ private <R> QueryResult<R> runTimestampedKeyQuery(
return result;
}
+ @SuppressWarnings("unchecked")
+ private <R> QueryResult<R> runTimestampedKeyWithHeadersQuery(
+ final Query<R> query,
+ final PositionBound positionBound,
+ final QueryConfig config
+ ) {
+ final QueryResult<R> result;
+ final TimestampedKeyWithHeadersQuery<K, V> typedKeyQuery =
(TimestampedKeyWithHeadersQuery<K, V>) query;
+
+ // Forward a raw byte-level KeyQuery to the wrapped store, propagating
skipCache so the caching
+ // layer can honor it; the result bytes are the serialized
ValueTimestampHeaders, which we
+ // deserialize below to recover value, timestamp, and headers.
+ KeyQuery<Bytes, byte[]> rawKeyQuery =
KeyQuery.withKey(serializeKey(typedKeyQuery.key(), internalContext.headers()));
+ if (typedKeyQuery.isSkipCache()) {
+ rawKeyQuery = rawKeyQuery.skipCache();
+ }
+ final QueryResult<byte[]> rawResult = wrapped().query(rawKeyQuery,
positionBound, config);
+ if (rawResult.isSuccess()) {
+ final Function<byte[], ValueTimestampHeaders<V>> deserializer =
StoreQueryUtils.deserializeValue(serdes, wrapped());
+ final ValueTimestampHeaders<V> valueTimestampHeaders =
deserializer.apply(rawResult.getResult());
+ // Surface the result as a ReadOnlyRecord (implemented by Record),
keeping the headers.
+ // A null wrapper means the key is absent or tombstoned, which we
surface as a null result.
+ final ReadOnlyRecord<K, V> record = valueTimestampHeaders == null
+ ? null
+ : new Record<>(
Review Comment:
`new Record<>(...)` throws `StreamsException` if
`valueTimestampHeaders.timestamp()` is negative, so a corrupted/unexpected
stored timestamp surfaces as an exception propagating out of `query()` rather
than as a failed `QueryResult`. Is a negative stored timestamp actually
impossible here? If so a brief comment would help; if not, consider mapping it
to a failed result instead.
##########
streams/src/test/java/org/apache/kafka/streams/state/internals/MeteredTimestampedKeyValueStoreWithHeadersTest.java:
##########
@@ -143,6 +151,61 @@ private void init() {
metered.init(context, metered);
}
+ // --- skipCache propagation: the metered handlers must forward
isSkipCache() onto the raw
+ // KeyQuery they build, otherwise the caching layer never sees it (it
was a no-op before). ---
+
+ @Test
+ public void shouldPropagateSkipCacheForKeyQuery() {
+ setUp();
+ init();
+
assertTrue(forwardedRawKeyQuery(KeyQuery.withKey(KEY).skipCache()).isSkipCache());
+ }
+
+ @Test
+ public void shouldNotSkipCacheForKeyQueryByDefault() {
+ setUp();
+ init();
+ assertFalse(forwardedRawKeyQuery(KeyQuery.withKey(KEY)).isSkipCache());
+ }
+
+ @Test
+ public void shouldPropagateSkipCacheForTimestampedKeyQuery() {
+ setUp();
+ init();
+
assertTrue(forwardedRawKeyQuery(TimestampedKeyQuery.withKey(KEY).skipCache()).isSkipCache());
+ }
+
+ @Test
+ public void shouldNotSkipCacheForTimestampedKeyQueryByDefault() {
+ setUp();
+ init();
+
assertFalse(forwardedRawKeyQuery(TimestampedKeyQuery.withKey(KEY)).isSkipCache());
+ }
+
+ @Test
+ public void shouldPropagateSkipCacheForTimestampedKeyWithHeadersQuery() {
+ setUp();
+ init();
+
assertTrue(forwardedRawKeyQuery(TimestampedKeyWithHeadersQuery.withKey(KEY).skipCache()).isSkipCache());
+ }
+
+ @Test
+ public void shouldNotSkipCacheForTimestampedKeyWithHeadersQueryByDefault()
{
+ setUp();
+ init();
+
assertFalse(forwardedRawKeyQuery(TimestampedKeyWithHeadersQuery.withKey(KEY)).isSkipCache());
+ }
+
+ @SuppressWarnings({"unchecked", "rawtypes"})
Review Comment:
`forwardedRawKeyQuery(...)` is copy-pasted verbatim into three test classes
(`MeteredKeyValueStoreTest`, `MeteredTimestampedKeyValueStoreTest`, and here).
Consider moving it to a shared test util to avoid the duplication.
##########
streams/src/main/java/org/apache/kafka/streams/state/internals/MeteredTimestampedKeyValueStoreWithHeaders.java:
##########
@@ -381,6 +390,45 @@ private <R> QueryResult<R> runTimestampedKeyQuery(
return result;
}
+ @SuppressWarnings("unchecked")
+ private <R> QueryResult<R> runTimestampedKeyWithHeadersQuery(
+ final Query<R> query,
+ final PositionBound positionBound,
+ final QueryConfig config
+ ) {
+ final QueryResult<R> result;
+ final TimestampedKeyWithHeadersQuery<K, V> typedKeyQuery =
(TimestampedKeyWithHeadersQuery<K, V>) query;
+
+ // Forward a raw byte-level KeyQuery to the wrapped store, propagating
skipCache so the caching
+ // layer can honor it; the result bytes are the serialized
ValueTimestampHeaders, which we
+ // deserialize below to recover value, timestamp, and headers.
+ KeyQuery<Bytes, byte[]> rawKeyQuery =
KeyQuery.withKey(serializeKey(typedKeyQuery.key(), internalContext.headers()));
+ if (typedKeyQuery.isSkipCache()) {
Review Comment:
The `if (typedKeyQuery.isSkipCache()) { rawKeyQuery =
rawKeyQuery.skipCache(); }` block is now repeated across ~5 handler methods
(here and in `MeteredKeyValueStore` / `MeteredTimestampedKeyValueStore`).
Consider extracting a small helper that builds the raw `KeyQuery` and applies
`skipCache`, to keep the propagation in one place.
--
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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
