aliehsaeedii commented on code in PR #22799:
URL: https://github.com/apache/kafka/pull/22799#discussion_r3568579371
##########
streams/src/main/java/org/apache/kafka/streams/state/internals/MeteredTimestampedWindowStoreWithHeaders.java:
##########
@@ -412,6 +464,87 @@ public Windowed<K> peekNextKey() {
}
}
+ /**
+ * Iterator backing {@link TimestampedWindowKeyWithHeadersQuery}: yields
each entry as a
+ * {@link ReadOnlyRecord} (implemented by {@link Record}) whose key is a
{@link Windowed} of the
+ * queried key and the entry's window, carrying the value, stored
event-time timestamp, and the
+ * stored headers, with the headers frozen so a caller cannot mutate the
read-only result.
+ *
+ * <p>A {@link ReadOnlyRecord} timestamp is contractually non-negative, so
an entry with a
+ * negative stored timestamp cannot be represented -- for example a {@code
WithHeaders} store
+ * built over a plain, non-timestamped window supplier surfaces entries
with
+ * {@code NO_TIMESTAMP} (-1). This mirrors the rule the point/range
headers queries apply. But
+ * because a lazily-evaluated iterator has already returned a successful
{@link QueryResult}
+ * before any entry is read, such an entry cannot be surfaced as a
query-level failure; it is
+ * instead reported by throwing a {@link StreamsException} while advancing
the iterator.
+ */
+ private class MeteredWindowStoreWithHeadersReadOnlyRecordIterator
+ implements ReadOnlyRecordIterator<Windowed<K>, V>, MeteredIterator {
+
+ private final WindowStoreIterator<byte[]> iter;
+ private final K key;
+ private final long startNs;
+ private final long startTimestampMs;
+
+ private MeteredWindowStoreWithHeadersReadOnlyRecordIterator(
+ final WindowStoreIterator<byte[]> iter,
+ final K key
+ ) {
+ this.iter = iter;
+ this.key = key;
+ 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>, V> next() {
+ final KeyValue<Long, byte[]> next = iter.next();
+ final long windowStart = next.key;
+ final ValueTimestampHeaders<V> valueTimestampHeaders =
deserializeValue(next.value);
+ final Headers headers = valueTimestampHeaders.headers();
+ if (valueTimestampHeaders.timestamp() < 0) {
+ throw new StreamsException(
Review Comment:
When `next()` throws here it doesn't close the iterator, so
`num-open-iterators` stays incremented until the caller closes it. The kv
`ReadOnlyRecord` iterator's Javadoc spells this out (callers must close in a
finally/try-with-resources even when `next()` throws) — add the same note to
this iterator's Javadoc.
##########
streams/src/main/java/org/apache/kafka/streams/state/internals/MeteredTimestampedWindowStoreWithHeaders.java:
##########
@@ -213,6 +224,47 @@ private <R> QueryResult<R> runWindowKeyQuery(
return queryResult;
}
+ /**
+ * Handles {@link TimestampedWindowKeyWithHeadersQuery} by forwarding a
raw byte-level
+ * {@link WindowKeyQuery} to the wrapped store and surfacing each entry as
a {@link ReadOnlyRecord}
+ * (via {@link Record}) whose key is a {@link Windowed} of the queried key
and the entry's window,
+ * carrying the stored value, event-time timestamp, and headers.
+ */
+ @SuppressWarnings("unchecked")
+ private <R> QueryResult<R> runTimestampedWindowKeyWithHeadersQuery(
+ final TimestampedWindowKeyWithHeadersQuery<K, V> query,
+ final PositionBound positionBound,
+ final QueryConfig config
+ ) {
+ final QueryResult<R> queryResult;
+
+ if (query.timeFrom().isPresent() && query.timeTo().isPresent()) {
Review Comment:
`timeFrom` and `timeTo` are always present (the only factory
`withKeyAndWindowStartRange` sets both via `Optional.of`), so this `else`
returning UNKNOWN_QUERY_TYPE is unreachable. Drop it, or keep it only for
parity with `runWindowKeyQuery`.
##########
streams/src/test/java/org/apache/kafka/streams/state/internals/MeteredTimestampedWindowStoreWithHeadersTest.java:
##########
@@ -484,4 +495,54 @@ public void
shouldUseHeadersFromValueToDeserializeKeyInBackwardFetchRange() {
verify(keyDeserializer).deserialize(any(), eq(HEADERS),
eq(KEY.getBytes()));
}
+
+ @Test
+ public void
shouldForwardWindowKeyQueryBoundsForTimestampedWindowKeyWithHeadersQuery() {
Review Comment:
The kv store test has `shouldLeaveIteratorOpenWhenNextThrowsAndNotClosed...`
for the leak when `next()` throws, but there's no window-store equivalent.
Consider adding one that checks `num-open-iterators` is 1 after the throw and 0
after `close()`.
--
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]
