aliehsaeedii commented on code in PR #22853:
URL: https://github.com/apache/kafka/pull/22853#discussion_r3637508069
##########
streams/src/main/java/org/apache/kafka/streams/state/internals/MeteredSessionStoreWithHeaders.java:
##########
@@ -183,21 +187,28 @@ public <R> QueryResult<R> query(
final PositionBound positionBound,
final QueryConfig config
) {
- final long start = config.isCollectExecutionInfo() ? System.nanoTime()
: -1L;
+ final long start = time.nanoseconds();
final QueryResult<R> result;
+ // Queries this store handles itself go through its serdes; anything
delegated to the wrapped
+ // store does not, so the two cases report execution info differently.
+ final boolean handledLocally;
if (query instanceof WindowRangeQuery) {
result = runRangeQuery((WindowRangeQuery<K, AGG>) query,
positionBound, config);
- if (config.isCollectExecutionInfo()) {
- result.addExecutionInfo(
- "Handled in " + getClass() + " with serdes " + serdes + "
in " + (time.nanoseconds() - start) + "ns");
- }
+ handledLocally = true;
+ } else if (query instanceof TimestampedWindowRangeWithHeadersQuery) {
+ result =
runTimestampedWindowRangeWithHeadersQuery((TimestampedWindowRangeWithHeadersQuery<K,
AGG>) query, positionBound, config);
+ handledLocally = true;
} else {
result = wrapped().query(query, positionBound, config);
- if (config.isCollectExecutionInfo()) {
- result.addExecutionInfo(
- "Handled in " + getClass() + " in " + (time.nanoseconds()
- start) + "ns");
- }
+ handledLocally = false;
+ }
+
+ if (config.isCollectExecutionInfo()) {
+ result.addExecutionInfo(
+ "Handled in " + getClass()
Review Comment:
No test asserts the new exec-info here: the `with serdes` suffix branch and
the clock now coming from `time.nanoseconds()` aren't covered (existing tests
only check the string contains the class name).
##########
streams/src/main/java/org/apache/kafka/streams/state/internals/MeteredTimestampedWindowStoreWithHeaders.java:
##########
@@ -152,17 +154,15 @@ public <R> QueryResult<R> query(
result =
runTimestampedWindowKeyWithHeadersQuery((TimestampedWindowKeyWithHeadersQuery<K,
V>) query, positionBound, config);
} else if (query instanceof WindowRangeQuery) {
result = runWindowRangeQuery((WindowRangeQuery<K,
ValueTimestampHeaders<V>>) query, positionBound, config);
+ } else if (query instanceof TimestampedWindowRangeWithHeadersQuery) {
+ result =
runTimestampedWindowRangeWithHeadersQuery((TimestampedWindowRangeWithHeadersQuery<K,
V>) query, positionBound, config);
} else {
result = wrapped().query(query, positionBound, config);
}
if (config.isCollectExecutionInfo()) {
- final String conversionType = isUnderlyingStoreTimestamped()
- ? "with conversion to ValueAndTimestamp"
- : "with extraction of plain values";
result.addExecutionInfo(
- "Handled in " + getClass() + " " + conversionType + " in "
- + (time.nanoseconds() - start) + "ns");
+ "Handled in " + getClass() + " in " + (time.nanoseconds() -
start) + "ns");
Review Comment:
This removes the `conversionType` detail (`with conversion to
ValueAndTimestamp` / `with extraction of plain values`) from the exec-info.
Intended? Note the sibling `MeteredSessionStoreWithHeaders.query()` (this same
PR) keeps a `with serdes` vs delegated distinction, so the two `*WithHeaders`
stores now emit different exec-info formats.
##########
streams/src/main/java/org/apache/kafka/streams/state/internals/MeteredSessionStoreWithHeaders.java:
##########
@@ -521,4 +573,71 @@ public Windowed<K> peekNextKey() {
return cachedNext.key;
}
}
+
+ /**
+ * Iterator backing the {@code withKey} form of {@link
TimestampedWindowRangeWithHeadersQuery}:
+ * yields each session as a {@link ReadOnlyRecord} (implemented by {@link
Record}) whose key is a
+ * {@link Windowed} of the deserialized key and the session's window,
carrying the aggregation and
+ * stored headers, with the headers frozen so a caller cannot mutate the
read-only result.
+ *
+ * <p>Unlike the window-store range query's iterator, there is no
negative-timestamp check here:
+ * {@link ReadOnlyRecord#timestamp()} is sourced from the session window's
end, which is validated
+ * non-negative when the window is constructed, so this iterator's {@code
next()} can never throw.
+ */
+ private class MeteredSessionWithHeadersReadOnlyRecordIterator
Review Comment:
Could we refactor (de-duplicate) the Metered*WithHeaders read-only-record
iterators
Problem:
Across the Metered*WithHeaders stores, each iterator hand-rolls the same
metering lifecycle. The same block of code is copy-pasted in
MeteredSessionStoreWithHeaders, MeteredTimestampedWindowStoreWithHeaders,
etc. The only part that genuinely differs is next(): raw key/value types,
value deserialization,
key derivation, timestamp source, and the negative-timestamp check (window)
vs none (session).
Could we have a shared generic iterator so that every store can
implement/extend it? We can do it as
a SEPARATE follow-up refactor PR, not folded into the feature PR.
##########
streams/src/main/java/org/apache/kafka/streams/state/internals/MeteredSessionStoreWithHeaders.java:
##########
@@ -183,21 +187,28 @@ public <R> QueryResult<R> query(
final PositionBound positionBound,
final QueryConfig config
) {
- final long start = config.isCollectExecutionInfo() ? System.nanoTime()
: -1L;
+ final long start = time.nanoseconds();
Review Comment:
Would be nice if you add a comment here (only in the PR - not int he code)
for the reviewer to know the reason for such changes. Thanks.
##########
streams/src/main/java/org/apache/kafka/streams/state/internals/MeteredSessionStoreWithHeaders.java:
##########
@@ -521,4 +573,71 @@ public Windowed<K> peekNextKey() {
return cachedNext.key;
}
}
+
+ /**
+ * Iterator backing the {@code withKey} form of {@link
TimestampedWindowRangeWithHeadersQuery}:
+ * yields each session as a {@link ReadOnlyRecord} (implemented by {@link
Record}) whose key is a
+ * {@link Windowed} of the deserialized key and the session's window,
carrying the aggregation and
+ * stored headers, with the headers frozen so a caller cannot mutate the
read-only result.
+ *
+ * <p>Unlike the window-store range query's iterator, there is no
negative-timestamp check here:
+ * {@link ReadOnlyRecord#timestamp()} is sourced from the session window's
end, which is validated
+ * non-negative when the window is constructed, so this iterator's {@code
next()} can never throw.
+ */
+ private class MeteredSessionWithHeadersReadOnlyRecordIterator
+ implements ReadOnlyRecordIterator<Windowed<K>, AGG>, MeteredIterator {
+
+ private final KeyValueIterator<Windowed<Bytes>, byte[]> iter;
+ private final long startNs;
+ private final long startTimestampMs;
+
+ private MeteredSessionWithHeadersReadOnlyRecordIterator(
+ final KeyValueIterator<Windowed<Bytes>, byte[]> iter
+ ) {
+ this.iter = iter;
+ this.startNs = time.nanoseconds();
+ this.startTimestampMs = time.milliseconds();
+ numOpenIterators.increment();
+ openIterators.add(this);
+ }
+
+ @Override
+ public long startTimestamp() {
+ return startTimestampMs;
+ }
+
+ @Override
+ public boolean hasNext() {
+ return iter.hasNext();
+ }
+
+ @Override
+ public ReadOnlyRecord<Windowed<K>, AGG> next() {
+ final KeyValue<Windowed<Bytes>, byte[]> next = iter.next();
+ final AggregationWithHeaders<AGG> aggregationWithHeaders =
deserializeValue(next.value);
+ final Headers headers = aggregationWithHeaders.headers();
Review Comment:
The sibling `MeteredSessionStoreWithHeadersIterator.next()` guards this as
`value != null ? value.headers() : new RecordHeaders()`; here
`aggregationWithHeaders.headers()` would NPE if the value ever deserializes to
null. Match the sibling for consistency.
##########
streams/src/main/java/org/apache/kafka/streams/state/internals/RocksDBSessionStoreWithHeaders.java:
##########
@@ -16,20 +16,14 @@
*/
package org.apache.kafka.streams.state.internals;
-import org.apache.kafka.streams.processor.StateStore;
-import org.apache.kafka.streams.query.Position;
-import org.apache.kafka.streams.query.PositionBound;
-import org.apache.kafka.streams.query.Query;
-import org.apache.kafka.streams.query.QueryConfig;
-import org.apache.kafka.streams.query.QueryResult;
import org.apache.kafka.streams.state.HeadersBytesStore;
/**
* RocksDB-backed session store with support for record headers.
* <p>
- * This store extends {@link RocksDBSessionStore} and returns
- * {@link QueryResult#forUnknownQueryType(Query, StateStore)} for all queries,
- * as IQv2 query handling is done at the metered layer.
+ * This store extends {@link RocksDBSessionStore} and relies on its inherited
IQv2 query handling
Review Comment:
Removing the query() override drops the explicit query-type allowlist;
scoping now relies on the `instanceof WindowStore`/`SessionStore` checks inside
`StoreQueryUtils`. Safe today, but a future handler dispatching on a broader
interface would reach this native header store and return raw header-format
bytes with no guard at this layer. Worth a note on why the implicit gating is
enough.
--
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]
