aliehsaeedii commented on code in PR #22666:
URL: https://github.com/apache/kafka/pull/22666#discussion_r3510830502
##########
streams/src/main/java/org/apache/kafka/streams/state/internals/MeteredTimestampedKeyValueStoreWithHeaders.java:
##########
@@ -381,6 +385,67 @@ 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());
+ if (valueTimestampHeaders != null &&
valueTimestampHeaders.timestamp() < 0) {
+ // The result is modeled as a Record, whose constructor
rejects negative timestamps. A
+ // negative stored timestamp cannot arise from the normal
record-driven flow (the PAPI
+ // Record a processor stores already forbids it), so it
indicates corrupted/unexpected
+ // store state; surface it as a failed result rather than
letting `new Record<>` throw
+ // out of query().
+ final QueryResult<ReadOnlyRecord<K, V>> failure =
QueryResult.forFailure(
+ FailureReason.STORE_EXCEPTION,
+ "Stored record for the queried key has a negative
timestamp ("
+ + valueTimestampHeaders.timestamp() + "); cannot
construct a ReadOnlyRecord.");
+ failure.setPosition(rawResult.getPosition());
Review Comment:
Minor/observability: the negative-timestamp failure builds a fresh
`QueryResult.forFailure(...)` and copies only the position. Every other exit of
this handler preserves the wrapped store's execution info — the success path
via `copyAndSubstituteDeserializedResult` (which copies
`rawResult.getExecutionInfo()`), and the `else` branch that returns `rawResult`
directly. So when `collectExecutionInfo` is enabled, this one failure path
silently drops the `"Handled in ..."` execution info that all sibling paths
keep. Consider copying `rawResult.getExecutionInfo()` onto `failure` (and/or
adding a note) for parity. Also worth noting:
`shouldCollectExecutionInfoWhenRequested` exercises a raw `KeyQuery` on the
wrapped store, not the typed `TimestampedKeyWithHeadersQuery` through the
metered handler, so neither the success nor this failure exec-info behavior of
the new query type is covered by a test.
##########
streams/src/main/java/org/apache/kafka/streams/state/internals/RocksDBTimestampedStoreWithHeaders.java:
##########
@@ -194,25 +190,4 @@ private void openFromTimestampedStore(final DBOptions
dbOptions,
}
}
- @SuppressWarnings("SynchronizeOnNonFinalField")
- @Override
- public <R> QueryResult<R> query(final Query<R> query,
- final PositionBound positionBound,
- final QueryConfig config) {
- final long start = config.isCollectExecutionInfo() ? System.nanoTime()
: -1L;
- final QueryResult<R> result;
-
- synchronized (position) {
- result = QueryResult.forUnknownQueryType(query, this);
-
- if (config.isCollectExecutionInfo()) {
- result.addExecutionInfo(
- "Handled in " + this.getClass() + " in " +
(System.nanoTime() - start) + "ns"
- );
- }
- result.setPosition(position.copy());
- }
- return result;
- }
-
-}
\ No newline at end of file
+}
Review Comment:
Have you checked which queries’ behavior changes when this method is
removed? The range query is affected (first thing comes to my mind- please
check other q types as well), which doesn’t seem correct, since this PR is only
adding key queries.
##########
streams/src/test/java/org/apache/kafka/streams/state/internals/TimestampedKeyValueStoreBuilderWithHeadersTest.java:
##########
@@ -586,247 +431,108 @@ public void
shouldReturnUnknownQueryTypeForKeyQueryOnHeadersStore(final boolean
@ParameterizedTest
@ValueSource(booleans = {true, false})
- public void shouldReturnUnknownQueryTypeForRangeQueryOnHeadersStore(final
boolean cachingEnabled) {
Review Comment:
Why did you delete this test? Nothing should be changed about `RangeQuery`
in this PR as it is only introducing key query.
--
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]
