aliehsaeedii commented on code in PR #22853: URL: https://github.com/apache/kafka/pull/22853#discussion_r3648004836
########## streams/src/main/java/org/apache/kafka/streams/query/TimestampedWindowRangeWithHeadersQuery.java: ########## @@ -0,0 +1,153 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.kafka.streams.query; + +import org.apache.kafka.common.annotation.InterfaceAudience; +import org.apache.kafka.common.annotation.InterfaceStability.Evolving; +import org.apache.kafka.streams.kstream.Windowed; +import org.apache.kafka.streams.processor.api.ReadOnlyRecord; +import org.apache.kafka.streams.state.ReadOnlyRecordIterator; +import org.apache.kafka.streams.state.SessionStoreWithHeaders; +import org.apache.kafka.streams.state.TimestampedWindowStoreWithHeaders; + +import java.time.Instant; +import java.util.Objects; +import java.util.Optional; + +/** + * Interactive query for retrieving records across a window-start range or for a single key's + * sessions, including their record headers, from a {@link TimestampedWindowStoreWithHeaders} or a + * {@link SessionStoreWithHeaders}. + * + * <p>This is the headers-aware parallel of {@link WindowRangeQuery}: it returns a + * {@link ReadOnlyRecordIterator} of {@link ReadOnlyRecord} elements, each carrying the windowed key, + * value, timestamp, and headers, whereas {@link WindowRangeQuery} returns a plain + * {@link org.apache.kafka.streams.state.KeyValueIterator} of values (no headers). Like + * {@link WindowRangeQuery}, this query has two mutually exclusive forms: + * + * <ul> + * <li>{@link #withWindowStartRange(Instant, Instant)} is handled by window stores: it retrieves + * every key whose window start falls within the closed range {@code [timeFrom, timeTo]}. Each + * element's key is a {@link Windowed} describing the record's window, and + * {@link ReadOnlyRecord#timestamp()} is the stored record's event-time. A stored event-time is + * contractually non-negative; if the backing store does not persist timestamps -- for example a + * {@code WithHeaders} store built over a plain window-store supplier that surfaces entries with + * {@code NO_TIMESTAMP} ({@code -1}) -- that entry cannot be represented, so advancing the + * returned {@link ReadOnlyRecordIterator} throws + * {@link org.apache.kafka.streams.errors.StreamsException} at that entry. Because iteration can + * therefore throw mid-stream, always close the returned iterator (for example with + * try-with-resources), even when a call to {@code next()} throws; otherwise the underlying store + * iterator leaks and the {@code num-open-iterators} metric stays incremented.</li> + * + * <li>{@link #withKey(Object)} is handled by session stores: it retrieves every session for the + * given key. Each element's key is a {@link Windowed} whose window is the session's window. + * Session aggregations carry no per-record event-time of their own, so + * {@link ReadOnlyRecord#timestamp()} is filled from the session window's (inclusive) end + * timestamp. That value is validated non-negative when the window is constructed, so -- unlike + * the window-store form above -- this form can never throw while iterating.</li> Review Comment: Worth adding that a session whose stored value deserializes to null comes back with `value() == null` here, while the window-store form throws for the same case. ########## streams/src/main/java/org/apache/kafka/streams/processor/api/ReadOnlyRecord.java: ########## @@ -55,10 +55,8 @@ public interface ReadOnlyRecord<K, V> { * The headers of the record. Never null. * * <p>The returned {@link Headers} is part of a read-only view and must not be mutated by - * callers. + * callers. Records served as IQv2 results from a state store have their headers frozen + * via RecordHeaders.setReadOnly(), so that any attempt to mutate them throws {@link IllegalStateException}. Review Comment: `setReadOnly()` only blocks add/remove — a caller can still change a header's `value()` bytes in place (`new RecordHeaders(headers)` is a shallow copy). Narrow this to say add/remove throw. ########## streams/src/test/java/org/apache/kafka/streams/state/internals/MeteredSessionStoreWithHeadersTest.java: ########## @@ -797,6 +801,198 @@ public void shouldFailWindowRangeQueryWithoutKey() { assertEquals(FailureReason.UNKNOWN_QUERY_TYPE, result.getFailureReason()); } + @SuppressWarnings("unchecked") + @Test + public void shouldHandleTimestampedWindowRangeWithHeadersQueryWithKey() { + setUp(); + init(); + + final Headers headers = new RecordHeaders(); + headers.add("key1", "value1".getBytes()); + final AggregationWithHeaders<String> valueAndHeaders = AggregationWithHeaders.make(VALUE, headers); + final AggregationWithHeadersSerializer<String> serializer = new AggregationWithHeadersSerializer<>(Serdes.String().serializer()); + final byte[] serializedValue = serializer.serialize(CHANGELOG_TOPIC, valueAndHeaders); + + // A non-degenerate session window (unlike the file's default WINDOWED_KEY_BYTES, whose + // SessionWindow(0, 0) would make a timestamp-from-window-end assertion trivially true). + final Windowed<Bytes> windowedKeyBytes = new Windowed<>(KEY_BYTES, new SessionWindow(START_TIMESTAMP, END_TIMESTAMP)); + final QueryResult<KeyValueIterator<Windowed<Bytes>, byte[]>> rawResult = + QueryResult.forResult(new KeyValueIteratorStub<>( + Collections.singleton(KeyValue.pair(windowedKeyBytes, serializedValue)).iterator())); + when(innerStore.query(any(), any(PositionBound.class), any(QueryConfig.class))) + .thenReturn((QueryResult) rawResult); + + final QueryResult<ReadOnlyRecordIterator<Windowed<String>, String>> result = store.query( + TimestampedWindowRangeWithHeadersQuery.<String, String>withKey(KEY), + PositionBound.unbounded(), + new QueryConfig(false)); + + assertTrue(result.isSuccess()); + try (ReadOnlyRecordIterator<Windowed<String>, String> iterator = result.getResult()) { + assertTrue(iterator.hasNext()); + final ReadOnlyRecord<Windowed<String>, String> record = iterator.next(); + assertEquals(KEY, record.key().key()); + assertEquals(START_TIMESTAMP, record.key().window().start()); + assertEquals(END_TIMESTAMP, record.key().window().end()); + assertEquals(VALUE, record.value()); + // The timestamp is sourced from the session window's end, not any per-record field + // (AggregationWithHeaders carries no timestamp of its own). + assertEquals(END_TIMESTAMP, record.timestamp()); + assertEquals(headers, record.headers()); + // returned headers are a read-only snapshot: neither add nor remove is allowed + assertThrows(IllegalStateException.class, () -> record.headers().add("x", new byte[0])); + assertThrows(IllegalStateException.class, () -> record.headers().remove("key1")); + assertFalse(iterator.hasNext()); + } + } + + @SuppressWarnings({"unchecked", "rawtypes"}) + @Test + public void shouldTolerateNullValueForTimestampedWindowRangeWithHeadersQuery() { + setUp(); + init(); + + // A value that deserializes to null must not NPE: the record surfaces a null value and empty + // headers, matching the sibling MeteredSessionStoreWithHeadersIterator.next(). + final Windowed<Bytes> windowedKeyBytes = new Windowed<>(KEY_BYTES, new SessionWindow(START_TIMESTAMP, END_TIMESTAMP)); + when(innerStore.query(any(), any(PositionBound.class), any(QueryConfig.class))) + .thenReturn((QueryResult) QueryResult.forResult(new KeyValueIteratorStub<>( + Collections.singleton(KeyValue.pair(windowedKeyBytes, (byte[]) null)).iterator()))); + + final QueryResult<ReadOnlyRecordIterator<Windowed<String>, String>> result = store.query( + TimestampedWindowRangeWithHeadersQuery.<String, String>withKey(KEY), + PositionBound.unbounded(), + new QueryConfig(false)); + + assertTrue(result.isSuccess()); + try (ReadOnlyRecordIterator<Windowed<String>, String> iterator = result.getResult()) { + assertTrue(iterator.hasNext()); + final ReadOnlyRecord<Windowed<String>, String> record = iterator.next(); + assertEquals(KEY, record.key().key()); + assertNull(record.value()); + assertEquals(END_TIMESTAMP, record.timestamp()); + assertEquals(new RecordHeaders(), record.headers()); + assertFalse(iterator.hasNext()); + } + } + + @Test + public void shouldRejectWithWindowStartRangeFormForTimestampedWindowRangeWithHeadersQuery() { + setUp(); + init(); + + final QueryResult<ReadOnlyRecordIterator<Windowed<String>, String>> result = store.query( + TimestampedWindowRangeWithHeadersQuery.<String, String>withWindowStartRange( + java.time.Instant.ofEpochMilli(0L), java.time.Instant.ofEpochMilli(0L)), + PositionBound.unbounded(), + new QueryConfig(false)); + + assertTrue(result.isFailure()); + assertEquals(FailureReason.UNKNOWN_QUERY_TYPE, result.getFailureReason()); + assertTrue( + result.getFailureMessage().contains("SessionStores only support TimestampedWindowRangeWithHeadersQuery.withKey"), + "unexpected message: " + result.getFailureMessage()); + } + + @SuppressWarnings({"unchecked", "rawtypes"}) + @Test + public void shouldForwardRawKeyQueryForTimestampedWindowRangeWithHeadersQuery() { + setUp(); + init(); + + when(innerStore.query(any(), any(PositionBound.class), any(QueryConfig.class))) + .thenReturn((QueryResult) QueryResult.forResult( + new KeyValueIteratorStub<>(Collections.<KeyValue<Windowed<Bytes>, byte[]>>emptyList().iterator()))); + + store.query( Review Comment: This opens an iterator via `query()` and never closes it, so `num-open-iterators` stays at 1. Take the result in try-with-resources. ########## streams/src/test/java/org/apache/kafka/streams/state/internals/MeteredTimestampedWindowStoreWithHeadersTest.java: ########## @@ -677,6 +680,212 @@ public KeyValue<Long, byte[]> next() { }; } + @Test + public void shouldPropagateWindowRangeBoundsForTimestampedWindowRangeWithHeadersQuery() { + setUp(); + store.init(context, store); + + final Instant timeFrom = Instant.ofEpochMilli(5); + final Instant timeTo = Instant.ofEpochMilli(100); + final WindowRangeQuery<?, ?> rawQuery = forwardedRawRangeQuery( + TimestampedWindowRangeWithHeadersQuery.<String, String>withWindowStartRange(timeFrom, timeTo)); + + // The typed query is translated into a raw byte-level WindowRangeQuery.withWindowStartRange + // with the same window-start range, forwarded to the wrapped store. + assertEquals(Optional.empty(), rawQuery.getKey()); + assertEquals(Optional.of(timeFrom), rawQuery.getTimeFrom()); + assertEquals(Optional.of(timeTo), rawQuery.getTimeTo()); + } + + @Test + public void shouldRejectWithKeyFormForTimestampedWindowRangeWithHeadersQuery() { + setUp(); + store.init(context, store); + + final QueryResult<ReadOnlyRecordIterator<Windowed<String>, String>> result = store.query( + TimestampedWindowRangeWithHeadersQuery.<String, String>withKey(KEY), + PositionBound.unbounded(), + new QueryConfig(false)); + + assertFalse(result.isSuccess()); + assertEquals(FailureReason.UNKNOWN_QUERY_TYPE, result.getFailureReason()); + assertTrue( + result.getFailureMessage().contains("WindowStores only supports TimestampedWindowRangeWithHeadersQuery.withWindowStartRange"), + "unexpected message: " + result.getFailureMessage()); + } + + @SuppressWarnings({"unchecked", "rawtypes"}) + @Test + public void shouldPropagateWrappedStoreFailureForTimestampedWindowRangeWithHeadersQuery() { + setUp(); + store.init(context, store); + when(innerStoreMock.query(any(), any(PositionBound.class), any(QueryConfig.class))) + .thenReturn((QueryResult) QueryResult.forFailure(FailureReason.STORE_EXCEPTION, "boom")); + + final QueryResult<ReadOnlyRecordIterator<Windowed<String>, String>> result = store.query( + TimestampedWindowRangeWithHeadersQuery.<String, String>withWindowStartRange( + Instant.ofEpochMilli(5), Instant.ofEpochMilli(100)), + PositionBound.unbounded(), + new QueryConfig(false)); + + assertFalse(result.isSuccess()); + assertEquals(FailureReason.STORE_EXCEPTION, result.getFailureReason()); + assertEquals("boom", result.getFailureMessage()); + } + + @SuppressWarnings({"unchecked", "rawtypes"}) + @Test + public void shouldTrackNumOpenIteratorsForTimestampedWindowRangeWithHeadersQuery() { + setUp(); + store.init(context, store); + when(innerStoreMock.query(any(), any(PositionBound.class), any(QueryConfig.class))) + .thenReturn((QueryResult) QueryResult.forResult(windowRangeIterator(List.of()))); + + final KafkaMetric openIterators = numOpenIteratorsMetric(); + assertEquals(0L, (Long) openIterators.metricValue()); + + final QueryResult<ReadOnlyRecordIterator<Windowed<String>, String>> result = store.query( + TimestampedWindowRangeWithHeadersQuery.<String, String>withWindowStartRange( + Instant.ofEpochMilli(5), Instant.ofEpochMilli(100)), + PositionBound.unbounded(), + new QueryConfig(false)); + assertTrue(result.isSuccess()); + + // The query's ReadOnlyRecordIterator registers itself on open and deregisters on close. + try (ReadOnlyRecordIterator<Windowed<String>, String> iterator = result.getResult()) { + assertEquals(1L, (Long) openIterators.metricValue()); + } + assertEquals(0L, (Long) openIterators.metricValue()); + } + + @SuppressWarnings({"unchecked", "rawtypes"}) + @Test + public void shouldDecrementOpenIteratorsTwiceWhenClosedTwiceForTimestampedWindowRangeWithHeadersQuery() { + setUp(); + store.init(context, store); + when(innerStoreMock.query(any(), any(PositionBound.class), any(QueryConfig.class))) + .thenReturn((QueryResult) QueryResult.forResult(windowRangeIterator(List.of()))); + + final KafkaMetric openIterators = numOpenIteratorsMetric(); + final ReadOnlyRecordIterator<Windowed<String>, String> iterator = store.query( + TimestampedWindowRangeWithHeadersQuery.<String, String>withWindowStartRange( + Instant.ofEpochMilli(5), Instant.ofEpochMilli(100)), + PositionBound.unbounded(), + new QueryConfig(false)).getResult(); + + assertEquals(1L, (Long) openIterators.metricValue()); + iterator.close(); + assertEquals(0L, (Long) openIterators.metricValue()); + // close() is intentionally not idempotent (matching the sibling metered iterators): each call + // decrements, so a repeated close drives the gauge below zero. Callers must close exactly once. + iterator.close(); + assertEquals(-1L, (Long) openIterators.metricValue()); + } + + @SuppressWarnings({"unchecked", "rawtypes"}) + @Test + public void shouldLeaveIteratorOpenWhenNextThrowsAndNotClosedForTimestampedWindowRangeWithHeadersQuery() { + setUp(); + store.init(context, store); + // A stored entry with a negative timestamp cannot be represented as a ReadOnlyRecord, so next() throws. + final byte[] negativeTimestampBytes = new ValueTimestampHeadersSerializer<>(new StringSerializer()) + .serialize("topic", ValueTimestampHeaders.make("value", -1L, HEADERS)); + final Windowed<Bytes> windowedKeyBytes = new Windowed<>(KEY_BYTES, new TimeWindow(5L, 5L + WINDOW_SIZE_MS)); + when(innerStoreMock.query(any(), any(PositionBound.class), any(QueryConfig.class))) + .thenReturn((QueryResult) QueryResult.forResult( + windowRangeIterator(List.of(KeyValue.pair(windowedKeyBytes, negativeTimestampBytes))))); + + final KafkaMetric openIterators = numOpenIteratorsMetric(); + final ReadOnlyRecordIterator<Windowed<String>, String> iterator = store.query( + TimestampedWindowRangeWithHeadersQuery.<String, String>withWindowStartRange( + Instant.ofEpochMilli(5), Instant.ofEpochMilli(100)), + PositionBound.unbounded(), + new QueryConfig(false)).getResult(); + + assertEquals(1L, (Long) openIterators.metricValue()); + assertThrows(StreamsException.class, iterator::next); + assertEquals(1L, (Long) openIterators.metricValue()); + iterator.close(); + assertEquals(0L, (Long) openIterators.metricValue()); + } + + @SuppressWarnings({"unchecked", "rawtypes"}) + @Test + public void shouldUseWindowFromRawResultForTimestampedWindowRangeWithHeadersQuery() { + setUp(); + store.init(context, store); + // Deliberately a window whose length differs from WINDOW_SIZE_MS, proving the returned + // Windowed<K> comes straight from the raw range result rather than being reconstructed from + // windowSizeMs (unlike the single-key point query, which only gets a window-start Long back). + final Windowed<Bytes> windowedKeyBytes = new Windowed<>(KEY_BYTES, new TimeWindow(1_000L, 5_000L)); + final KeyValue<Windowed<Bytes>, byte[]> testData = KeyValue.pair(windowedKeyBytes, VALUE_TIMESTAMP_HEADERS_BYTES); + when(innerStoreMock.query(any(), any(PositionBound.class), any(QueryConfig.class))) + .thenReturn((QueryResult) QueryResult.forResult(windowRangeIterator(List.of(testData)))); + + final QueryResult<ReadOnlyRecordIterator<Windowed<String>, String>> result = store.query( + TimestampedWindowRangeWithHeadersQuery.<String, String>withWindowStartRange( + Instant.ofEpochMilli(0), Instant.ofEpochMilli(10_000)), + PositionBound.unbounded(), + new QueryConfig(false)); + + assertTrue(result.isSuccess()); + try (ReadOnlyRecordIterator<Windowed<String>, String> iterator = result.getResult()) { + assertTrue(iterator.hasNext()); + final ReadOnlyRecord<Windowed<String>, String> record = iterator.next(); + assertEquals(KEY, record.key().key()); + assertEquals(1_000L, record.key().window().start()); + assertEquals(5_000L, record.key().window().end()); + assertEquals("value", record.value()); + assertEquals(TIMESTAMP, record.timestamp()); + assertEquals(HEADERS, record.headers()); + // returned headers are a read-only snapshot: neither add nor remove is allowed + assertThrows(IllegalStateException.class, () -> record.headers().add("x", new byte[0])); + assertThrows(IllegalStateException.class, () -> record.headers().remove("header-key")); + assertFalse(iterator.hasNext()); + } + } + + @SuppressWarnings({"unchecked", "rawtypes"}) + @Test + public void shouldThrowForNullStoredValueForTimestampedWindowRangeWithHeadersQuery() { + setUp(); + store.init(context, store); + + // A ReadOnlyRecord carries the stored event-time; a value that deserializes to null has none, + // so the entry cannot be represented and next() throws (rather than NPE-ing on the null value). + final Windowed<Bytes> windowedKeyBytes = new Windowed<>(KEY_BYTES, new TimeWindow(1_000L, 5_000L)); + when(innerStoreMock.query(any(), any(PositionBound.class), any(QueryConfig.class))) + .thenReturn((QueryResult) QueryResult.forResult( + windowRangeIterator(List.of(KeyValue.pair(windowedKeyBytes, (byte[]) null))))); + + final QueryResult<ReadOnlyRecordIterator<Windowed<String>, String>> result = store.query( + TimestampedWindowRangeWithHeadersQuery.<String, String>withWindowStartRange( + Instant.ofEpochMilli(0), Instant.ofEpochMilli(10_000)), + PositionBound.unbounded(), + new QueryConfig(false)); + + assertTrue(result.isSuccess()); + try (ReadOnlyRecordIterator<Windowed<String>, String> iterator = result.getResult()) { + assertTrue(iterator.hasNext()); + final StreamsException exception = assertThrows(StreamsException.class, iterator::next); + assertTrue(exception.getMessage().contains("its value is null"), exception.getMessage()); + } + } + + @SuppressWarnings({"unchecked", "rawtypes"}) + private WindowRangeQuery<?, ?> forwardedRawRangeQuery(final Query<?> query) { + when(innerStoreMock.query(any(), any(PositionBound.class), any(QueryConfig.class))) + .thenReturn((QueryResult) QueryResult.forResult(windowRangeIterator(List.of()))); + store.query(query, PositionBound.unbounded(), new QueryConfig(false)); Review Comment: Same here — this query's iterator is never closed. Close it before returning the captured raw 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]
