aliehsaeedii commented on code in PR #22853:
URL: https://github.com/apache/kafka/pull/22853#discussion_r3643287864
##########
streams/src/test/java/org/apache/kafka/streams/state/internals/RocksDBSessionStoreWithHeadersTest.java:
##########
@@ -83,15 +85,29 @@ public void tearDown() {
}
@Test
- public void shouldReturnUnknownQueryTypeForWindowRangeQuery() {
- final WindowRangeQuery<Bytes, byte[]> query = WindowRangeQuery.withKey(
- new Bytes("test-key".getBytes())
- );
+ public void shouldHandleWindowRangeQuery() {
+ // KIP-1356: the withKey form of the headers-aware
TimestampedWindowRangeWithHeadersQuery
+ // forwards a raw WindowRangeQuery to this native store, so enable
WindowRangeQuery via the
+ // inherited RocksDBSessionStore handling (StoreQueryUtils), returning
the raw stored
+ // header-format bytes (previously UNKNOWN_QUERY_TYPE for every
query). This also fixes the
+ // same pre-existing gap for the plain (non-headers)
WindowRangeQuery.withKey.
+ final Bytes key = new Bytes("test-key".getBytes());
+ final byte[] storedBytes = "headers+aggregation".getBytes();
+ final Windowed<Bytes> windowedKey = new Windowed<>(key, new
SessionWindow(0L, 1_000L));
+ sessionStore.put(windowedKey, storedBytes);
+
+ final WindowRangeQuery<Bytes, byte[]> query =
WindowRangeQuery.withKey(key);
final QueryResult<KeyValueIterator<Windowed<Bytes>, byte[]>> result =
sessionStore.query(query, PositionBound.unbounded(), new
QueryConfig(false));
- assertFalse(result.isSuccess());
- assertEquals(FailureReason.UNKNOWN_QUERY_TYPE,
result.getFailureReason());
+ assertTrue(result.isSuccess(), "Expected WindowRangeQuery to succeed");
+ try (KeyValueIterator<Windowed<Bytes>, byte[]> iterator =
result.getResult()) {
Review Comment:
The two exec-info tests just below
(`shouldCollectExecutionInfoWhenRequested` /
`shouldNotCollectExecutionInfoWhenNotRequested`) run the same
`WindowRangeQuery.withKey` but never close the result. That query used to fail
with no iterator; now that the override is gone it succeeds and returns an open
RocksDB iterator that leaks. Wrap them in try-with-resources like this test
does.
##########
streams/src/test/java/org/apache/kafka/streams/state/internals/RocksDBTimeOrderedSessionStoreWithHeadersTest.java:
##########
@@ -83,15 +85,29 @@ public void tearDown() {
}
@Test
- public void shouldReturnUnknownQueryTypeForWindowRangeQuery() {
- final WindowRangeQuery<Bytes, byte[]> query = WindowRangeQuery.withKey(
- new Bytes("test-key".getBytes())
- );
+ public void shouldHandleWindowRangeQuery() {
+ // KIP-1356: the withKey form of the headers-aware
TimestampedWindowRangeWithHeadersQuery
+ // forwards a raw WindowRangeQuery to this native store, so enable
WindowRangeQuery via the
+ // inherited RocksDBTimeOrderedSessionStore handling
(StoreQueryUtils), returning the raw
+ // stored header-format bytes (previously UNKNOWN_QUERY_TYPE for every
query). This also fixes
+ // the same pre-existing gap for the plain (non-headers)
WindowRangeQuery.withKey.
+ final Bytes key = new Bytes("test-key".getBytes());
+ final byte[] storedBytes = "headers+aggregation".getBytes();
+ final Windowed<Bytes> windowedKey = new Windowed<>(key, new
SessionWindow(0L, 1_000L));
+ sessionStore.put(windowedKey, storedBytes);
+
+ final WindowRangeQuery<Bytes, byte[]> query =
WindowRangeQuery.withKey(key);
final QueryResult<KeyValueIterator<Windowed<Bytes>, byte[]>> result =
sessionStore.query(query, PositionBound.unbounded(), new
QueryConfig(false));
- assertFalse(result.isSuccess());
- assertEquals(FailureReason.UNKNOWN_QUERY_TYPE,
result.getFailureReason());
+ assertTrue(result.isSuccess(), "Expected WindowRangeQuery to succeed");
+ try (KeyValueIterator<Windowed<Bytes>, byte[]> iterator =
result.getResult()) {
Review Comment:
Same as the session-store test: the exec-info tests below still run
`WindowRangeQuery.withKey` and drop the result unclosed. Now that this succeeds
instead of failing, it leaks an open iterator — wrap them in try-with-resources.
--
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]
