This is an automated email from the ASF dual-hosted git repository.
frankvicky pushed a commit to branch trunk
in repository https://gitbox.apache.org/repos/asf/kafka.git
The following commit(s) were added to refs/heads/trunk by this push:
new 209a1b6e39a KAFKA-20328 : [1/N] headers-aware api test coverage
(#22196)
209a1b6e39a is described below
commit 209a1b6e39a535453d302ca1306d24eded62bfda
Author: Murali Basani <[email protected]>
AuthorDate: Wed Jul 29 11:29:06 2026 +0200
KAFKA-20328 : [1/N] headers-aware api test coverage (#22196)
Ref : https://issues.apache.org/jira/browse/KAFKA-20328
After audit, found a few gaps and also considered the work of 1271 kip
This is first PR. Next PRs will be on Decorator/forwarder coverage —
AbstractReadWriteDecorator and AbstractReadX related and other
integration tests
This PR :
- new tests for PlainToHeadersWindowStoreIteratorAdapterTest,
RocksDBMigratingWindowStoreWithHeaders and
TimestampedToHeadersStoreAdapter (considered deleted files in
https://github.com/apache/kafka/pull/21830)
- Update StoresTest with coverage for all 6 *WithHeaders factory methods
on Stores
- New test : DslStoreFormatTest cases for HEADERS enum
- New contract tests for TimestampedKeyValueStoreWithHeaders,
TimestampedWindowStoreWithHeaders and SessionStoreWithHeaders (with put,
get, range, fetch, and backwardFetch, etc)
Reviewers: Alieh Saeedi <[email protected]>, TengYao Chi
<[email protected]>
---
.../apache/kafka/streams/DslStoreFormatTest.java | 57 +++
.../state/SessionStoreWithHeadersContractTest.java | 222 ++++++++++++
.../org/apache/kafka/streams/state/StoresTest.java | 128 +++++++
...tampedKeyValueStoreWithHeadersContractTest.java | 198 +++++++++++
...estampedWindowStoreWithHeadersContractTest.java | 225 ++++++++++++
...ainToHeadersWindowStoreIteratorAdapterTest.java | 114 ++++++
...RocksDBMigratingWindowStoreWithHeadersTest.java | 190 ++++++++++
.../TimestampedToHeadersStoreAdapterTest.java | 390 +++++++++++++++++++++
8 files changed, 1524 insertions(+)
diff --git
a/streams/src/test/java/org/apache/kafka/streams/DslStoreFormatTest.java
b/streams/src/test/java/org/apache/kafka/streams/DslStoreFormatTest.java
new file mode 100644
index 00000000000..f3c8ca09c98
--- /dev/null
+++ b/streams/src/test/java/org/apache/kafka/streams/DslStoreFormatTest.java
@@ -0,0 +1,57 @@
+/*
+ * 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;
+
+import org.junit.jupiter.api.Test;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+
+public class DslStoreFormatTest {
+
+ @Test
+ public void shouldResolvePlain() {
+ assertEquals(DslStoreFormat.PLAIN, DslStoreFormat.of("PLAIN"));
+ }
+
+ @Test
+ public void shouldResolveTimestamped() {
+ assertEquals(DslStoreFormat.TIMESTAMPED,
DslStoreFormat.of("TIMESTAMPED"));
+ }
+
+ @Test
+ public void shouldResolveHeaders() {
+ assertEquals(DslStoreFormat.HEADERS, DslStoreFormat.of("HEADERS"));
+ }
+
+ @Test
+ public void shouldResolveHeadersCaseInsensitively() {
+ assertEquals(DslStoreFormat.HEADERS, DslStoreFormat.of("headers"));
+ assertEquals(DslStoreFormat.HEADERS, DslStoreFormat.of("Headers"));
+ assertEquals(DslStoreFormat.HEADERS, DslStoreFormat.of("hEaDeRs"));
+ }
+
+ @Test
+ public void shouldExposeHeadersNameField() {
+ assertEquals("HEADERS", DslStoreFormat.HEADERS.name);
+ }
+
+ @Test
+ public void shouldThrowOnInvalidInput() {
+ assertThrows(IllegalArgumentException.class, () ->
DslStoreFormat.of("not-a-format"));
+ }
+}
diff --git
a/streams/src/test/java/org/apache/kafka/streams/state/SessionStoreWithHeadersContractTest.java
b/streams/src/test/java/org/apache/kafka/streams/state/SessionStoreWithHeadersContractTest.java
new file mode 100644
index 00000000000..309fabc1c9c
--- /dev/null
+++
b/streams/src/test/java/org/apache/kafka/streams/state/SessionStoreWithHeadersContractTest.java
@@ -0,0 +1,222 @@
+/*
+ * 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.state;
+
+import org.apache.kafka.common.header.Headers;
+import org.apache.kafka.common.header.internals.RecordHeader;
+import org.apache.kafka.common.header.internals.RecordHeaders;
+import org.apache.kafka.common.serialization.Serdes;
+import org.apache.kafka.streams.KeyValue;
+import org.apache.kafka.streams.StreamsConfig;
+import org.apache.kafka.streams.kstream.Windowed;
+import org.apache.kafka.streams.kstream.internals.SessionWindow;
+import org.apache.kafka.test.InternalMockProcessorContext;
+import org.apache.kafka.test.StreamsTestUtils;
+import org.apache.kafka.test.TestUtils;
+
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+import java.io.File;
+import java.time.Duration;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Properties;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertNull;
+
+/**
+ * Contract tests for {@link SessionStoreWithHeaders}.
+ */
+public class SessionStoreWithHeadersContractTest {
+
+ private static final long RETENTION = 10_000L;
+
+ private SessionStoreWithHeaders<String, String> store;
+ private InternalMockProcessorContext<String, String> context;
+
+ @BeforeEach
+ public void setUp() {
+ final File dir = TestUtils.tempDirectory();
+ final Properties props = StreamsTestUtils.getStreamsConfig();
+ context = new InternalMockProcessorContext<>(
+ dir,
+ Serdes.String(),
+ Serdes.String(),
+ new StreamsConfig(props)
+ );
+ store = Stores.sessionStoreWithHeadersBuilder(
+ Stores.inMemorySessionStore("contract-session-store",
Duration.ofMillis(RETENTION)),
+ Serdes.String(),
+ Serdes.String()
+ ).withLoggingDisabled().withCachingDisabled().build();
+ store.init(context, store);
+ }
+
+ @AfterEach
+ public void tearDown() {
+ if (store != null) {
+ store.close();
+ }
+ }
+
+ @Test
+ public void shouldRoundTripAggregationAndHeadersViaPutAndFetchSession() {
+ final Headers headers = headersWith("schema-id", "42");
+ final Windowed<String> key = windowed("k", 100L, 200L);
+
+ store.put(key, AggregationWithHeaders.make("agg", headers));
+
+ final AggregationWithHeaders<String> result = store.fetchSession("k",
100L, 200L);
+ assertEquals("agg", result.aggregation());
+ assertEquals(headers, result.headers());
+ }
+
+ @Test
+ public void shouldReturnNullForMissingSession() {
+ assertNull(store.fetchSession("missing", 0L, 10L));
+ }
+
+ @Test
+ public void shouldTreatNullAggregationAsTombstone() {
+ final Headers headers = headersWith("h", "v");
+ final Windowed<String> key = windowed("k", 100L, 200L);
+
+ store.put(key, AggregationWithHeaders.make("agg", headers));
+ assertEquals("agg", store.fetchSession("k", 100L, 200L).aggregation());
+
+ store.put(key, null);
+ assertNull(store.fetchSession("k", 100L, 200L));
+ }
+
+ @Test
+ public void shouldRemoveSessionByWindowedKey() {
+ final Headers headers = headersWith("h", "v");
+ final Windowed<String> key = windowed("k", 100L, 200L);
+
+ store.put(key, AggregationWithHeaders.make("agg", headers));
+ store.remove(key);
+
+ assertNull(store.fetchSession("k", 100L, 200L));
+ }
+
+ @Test
+ public void shouldPreserveHeadersAcrossFetchByKey() {
+ final Headers h1 = headersWith("id", "1");
+ final Headers h2 = headersWith("id", "2");
+ final Headers h3 = headersWith("id", "3");
+
+ store.put(windowed("k", 100L, 150L), AggregationWithHeaders.make("a1",
h1));
+ store.put(windowed("k", 200L, 250L), AggregationWithHeaders.make("a2",
h2));
+ store.put(windowed("k", 300L, 350L), AggregationWithHeaders.make("a3",
h3));
+
+ final List<KeyValue<Long, Headers>> collected = new ArrayList<>();
+ try (KeyValueIterator<Windowed<String>,
AggregationWithHeaders<String>> it = store.fetch("k")) {
+ while (it.hasNext()) {
+ final KeyValue<Windowed<String>,
AggregationWithHeaders<String>> next = it.next();
+ collected.add(KeyValue.pair(next.key.window().start(),
next.value.headers()));
+ }
+ }
+
+ assertEquals(3, collected.size());
+ assertEquals(100L, collected.get(0).key.longValue());
+ assertEquals(h1, collected.get(0).value);
+ assertEquals(200L, collected.get(1).key.longValue());
+ assertEquals(h2, collected.get(1).value);
+ assertEquals(300L, collected.get(2).key.longValue());
+ assertEquals(h3, collected.get(2).value);
+ }
+
+ @Test
+ public void shouldPreserveHeadersAcrossFindSessions() {
+ final Headers h1 = headersWith("id", "early");
+ final Headers h2 = headersWith("id", "late");
+
+ store.put(windowed("k", 100L, 150L), AggregationWithHeaders.make("a1",
h1));
+ store.put(windowed("k", 200L, 250L), AggregationWithHeaders.make("a2",
h2));
+
+ final List<KeyValue<Long, Headers>> collected = new ArrayList<>();
+ try (KeyValueIterator<Windowed<String>,
AggregationWithHeaders<String>> it =
+ store.findSessions("k", 100L, 250L)) {
+ while (it.hasNext()) {
+ final KeyValue<Windowed<String>,
AggregationWithHeaders<String>> next = it.next();
+ collected.add(KeyValue.pair(next.key.window().start(),
next.value.headers()));
+ }
+ }
+
+ assertEquals(2, collected.size());
+ assertEquals(100L, collected.get(0).key.longValue());
+ assertEquals(h1, collected.get(0).value);
+ assertEquals(200L, collected.get(1).key.longValue());
+ assertEquals(h2, collected.get(1).value);
+ }
+
+ @Test
+ public void shouldPreserveHeadersAcrossBackwardFindSessions() {
+ final Headers h1 = headersWith("id", "early");
+ final Headers h2 = headersWith("id", "late");
+
+ store.put(windowed("k", 100L, 150L), AggregationWithHeaders.make("a1",
h1));
+ store.put(windowed("k", 200L, 250L), AggregationWithHeaders.make("a2",
h2));
+
+ final List<KeyValue<Long, Headers>> collected = new ArrayList<>();
+ try (KeyValueIterator<Windowed<String>,
AggregationWithHeaders<String>> it =
+ store.backwardFindSessions("k", 100L, 250L)) {
+ while (it.hasNext()) {
+ final KeyValue<Windowed<String>,
AggregationWithHeaders<String>> next = it.next();
+ collected.add(KeyValue.pair(next.key.window().start(),
next.value.headers()));
+ }
+ }
+
+ assertEquals(2, collected.size());
+ assertEquals(200L, collected.get(0).key.longValue());
+ assertEquals(h2, collected.get(0).value);
+ assertEquals(100L, collected.get(1).key.longValue());
+ assertEquals(h1, collected.get(1).value);
+ }
+
+ @Test
+ public void shouldReturnEmptyIteratorForMissingKey() {
+ try (KeyValueIterator<Windowed<String>,
AggregationWithHeaders<String>> it = store.fetch("missing")) {
+ assertFalse(it.hasNext());
+ }
+ }
+
+ @Test
+ public void shouldPreserveEmptyHeaders() {
+ final Windowed<String> key = windowed("k", 100L, 200L);
+ store.put(key, AggregationWithHeaders.make("agg", new
RecordHeaders()));
+
+ final AggregationWithHeaders<String> result = store.fetchSession("k",
100L, 200L);
+ assertEquals("agg", result.aggregation());
+ assertEquals(new RecordHeaders(), result.headers());
+ assertEquals(0, result.headers().toArray().length);
+ }
+
+ private static Windowed<String> windowed(final String key, final long
start, final long end) {
+ return new Windowed<>(key, new SessionWindow(start, end));
+ }
+
+ private static Headers headersWith(final String key, final String value) {
+ final Headers headers = new RecordHeaders();
+ headers.add(new RecordHeader(key, value.getBytes()));
+ return headers;
+ }
+}
diff --git
a/streams/src/test/java/org/apache/kafka/streams/state/StoresTest.java
b/streams/src/test/java/org/apache/kafka/streams/state/StoresTest.java
index d5c588b59ee..0c5c7e41a27 100644
--- a/streams/src/test/java/org/apache/kafka/streams/state/StoresTest.java
+++ b/streams/src/test/java/org/apache/kafka/streams/state/StoresTest.java
@@ -324,4 +324,132 @@ public class StoresTest {
).build();
assertThat(store, not(nullValue()));
}
+
+ @Test
+ public void
shouldThrowIfPersistentTimestampedKeyValueStoreWithHeadersNameIsNull() {
+ final Exception e = assertThrows(NullPointerException.class,
+ () -> Stores.persistentTimestampedKeyValueStoreWithHeaders(null));
+ assertEquals("name cannot be null", e.getMessage());
+ }
+
+ @Test
+ public void
shouldThrowIfPersistentTimestampedWindowStoreWithHeadersNameIsNull() {
+ final Exception e = assertThrows(NullPointerException.class,
+ () -> Stores.persistentTimestampedWindowStoreWithHeaders(null,
ZERO, ZERO, false));
+ assertEquals("name cannot be null", e.getMessage());
+ }
+
+ @Test
+ public void
shouldThrowIfPersistentTimestampedWindowStoreWithHeadersRetentionPeriodIsNegative()
{
+ final Exception e = assertThrows(IllegalArgumentException.class,
+ () ->
Stores.persistentTimestampedWindowStoreWithHeaders("anyName", ofMillis(-1L),
ZERO, false));
+ assertEquals("retentionPeriod cannot be negative", e.getMessage());
+ }
+
+ @Test
+ public void
shouldThrowIfPersistentTimestampedWindowStoreWithHeadersWindowSizeIsNegative() {
+ final Exception e = assertThrows(IllegalArgumentException.class,
+ () ->
Stores.persistentTimestampedWindowStoreWithHeaders("anyName", ofMillis(0L),
ofMillis(-1L), false));
+ assertEquals("windowSize cannot be negative", e.getMessage());
+ }
+
+ @Test
+ public void shouldThrowIfPersistentSessionStoreWithHeadersNameIsNull() {
+ final Exception e = assertThrows(NullPointerException.class,
+ () -> Stores.persistentSessionStoreWithHeaders(null, ofMillis(0)));
+ assertEquals("name cannot be null", e.getMessage());
+ }
+
+ @Test
+ public void
shouldThrowIfPersistentSessionStoreWithHeadersRetentionPeriodIsNegative() {
+ final Exception e = assertThrows(IllegalArgumentException.class,
+ () -> Stores.persistentSessionStoreWithHeaders("anyName",
ofMillis(-1)));
+ assertEquals("retentionPeriod cannot be negative", e.getMessage());
+ }
+
+ @Test
+ public void
shouldThrowIfSupplierIsNullForTimestampedKeyValueStoreWithHeadersBuilder() {
+ final Exception e = assertThrows(NullPointerException.class,
+ () -> Stores.timestampedKeyValueStoreWithHeadersBuilder(null,
Serdes.String(), Serdes.String()));
+ assertEquals("supplier cannot be null", e.getMessage());
+ }
+
+ @Test
+ public void
shouldThrowIfSupplierIsNullForTimestampedWindowStoreWithHeadersBuilder() {
+ final Exception e = assertThrows(NullPointerException.class,
+ () -> Stores.timestampedWindowStoreWithHeadersBuilder(null,
Serdes.String(), Serdes.String()));
+ assertEquals("supplier cannot be null", e.getMessage());
+ }
+
+ @Test
+ public void shouldThrowIfSupplierIsNullForSessionStoreWithHeadersBuilder()
{
+ final Exception e = assertThrows(NullPointerException.class,
+ () -> Stores.sessionStoreWithHeadersBuilder(null, Serdes.String(),
Serdes.String()));
+ assertEquals("supplier cannot be null", e.getMessage());
+ }
+
+ @Test
+ public void
shouldCreatePersistentTimestampedKeyValueStoreWithHeadersSupplier() {
+ final KeyValueBytesStoreSupplier supplier =
Stores.persistentTimestampedKeyValueStoreWithHeaders("store");
+ assertThat(supplier.name(), equalTo("store"));
+ assertThat(supplier.metricsScope(), equalTo("rocksdb"));
+ assertThat(supplier.get(), not(nullValue()));
+ assertThat(supplier.get().persistent(), equalTo(true));
+ }
+
+ @Test
+ public void
shouldCreatePersistentTimestampedWindowStoreWithHeadersSupplier() {
+ final WindowBytesStoreSupplier supplier =
+ Stores.persistentTimestampedWindowStoreWithHeaders("store",
ofMillis(10L), ofMillis(5L), false);
+ assertThat(supplier.name(), equalTo("store"));
+ assertThat(supplier.windowSize(), equalTo(5L));
+ assertThat(supplier.retentionPeriod(), equalTo(10L));
+ assertThat(supplier.retainDuplicates(), equalTo(false));
+ assertThat(supplier.get(), not(nullValue()));
+ assertThat(supplier.get().persistent(), equalTo(true));
+ }
+
+ @Test
+ public void shouldCreatePersistentSessionStoreWithHeadersSupplier() {
+ final SessionBytesStoreSupplier supplier =
+ Stores.persistentSessionStoreWithHeaders("store", ofMillis(100L));
+ assertThat(supplier.name(), equalTo("store"));
+ assertThat(supplier.metricsScope(), equalTo("rocksdb-session"));
+ assertThat(supplier.retentionPeriod(), equalTo(100L));
+ assertThat(supplier.get(), not(nullValue()));
+ assertThat(supplier.get().persistent(), equalTo(true));
+ }
+
+ @Test
+ public void shouldBuildTimestampedKeyValueStoreWithHeaders() {
+ final TimestampedKeyValueStoreWithHeaders<String, String> store =
+ Stores.timestampedKeyValueStoreWithHeadersBuilder(
+ Stores.inMemoryKeyValueStore("name"),
+ Serdes.String(),
+ Serdes.String()
+ ).withLoggingDisabled().build();
+ assertThat(store, not(nullValue()));
+ }
+
+ @Test
+ public void shouldBuildTimestampedWindowStoreWithHeaders() {
+ final TimestampedWindowStoreWithHeaders<String, String> store =
+ Stores.timestampedWindowStoreWithHeadersBuilder(
+ Stores.inMemoryWindowStore("store", ofMillis(10L),
ofMillis(5L), false),
+ Serdes.String(),
+ Serdes.String()
+ ).withLoggingDisabled().build();
+ assertThat(store, not(nullValue()));
+ }
+
+ @Test
+ public void shouldBuildSessionStoreWithHeaders() {
+ final SessionStoreWithHeaders<String, String> store =
+ Stores.sessionStoreWithHeadersBuilder(
+ Stores.inMemorySessionStore("name", ofMillis(100L)),
+ Serdes.String(),
+ Serdes.String()
+ ).withLoggingDisabled().build();
+ assertThat(store, not(nullValue()));
+ }
}
diff --git
a/streams/src/test/java/org/apache/kafka/streams/state/TimestampedKeyValueStoreWithHeadersContractTest.java
b/streams/src/test/java/org/apache/kafka/streams/state/TimestampedKeyValueStoreWithHeadersContractTest.java
new file mode 100644
index 00000000000..7f2ead2b08f
--- /dev/null
+++
b/streams/src/test/java/org/apache/kafka/streams/state/TimestampedKeyValueStoreWithHeadersContractTest.java
@@ -0,0 +1,198 @@
+/*
+ * 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.state;
+
+import org.apache.kafka.common.header.Headers;
+import org.apache.kafka.common.header.internals.RecordHeader;
+import org.apache.kafka.common.header.internals.RecordHeaders;
+import org.apache.kafka.common.serialization.Serdes;
+import org.apache.kafka.streams.KeyValue;
+import org.apache.kafka.streams.StreamsConfig;
+import org.apache.kafka.test.InternalMockProcessorContext;
+import org.apache.kafka.test.StreamsTestUtils;
+import org.apache.kafka.test.TestUtils;
+
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+import java.io.File;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Properties;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertNull;
+
+/**
+ * Contract tests for {@link TimestampedKeyValueStoreWithHeaders}.
+ */
+public class TimestampedKeyValueStoreWithHeadersContractTest {
+
+ private TimestampedKeyValueStoreWithHeaders<String, String> store;
+ private InternalMockProcessorContext<String, String> context;
+
+ @BeforeEach
+ public void setUp() {
+ final File dir = TestUtils.tempDirectory();
+ final Properties props = StreamsTestUtils.getStreamsConfig();
+ context = new InternalMockProcessorContext<>(
+ dir,
+ Serdes.String(),
+ Serdes.String(),
+ new StreamsConfig(props)
+ );
+ store = Stores.timestampedKeyValueStoreWithHeadersBuilder(
+ Stores.inMemoryKeyValueStore("contract-store"),
+ Serdes.String(),
+ Serdes.String()
+ ).withLoggingDisabled().withCachingDisabled().build();
+ store.init(context, store);
+ }
+
+ @AfterEach
+ public void tearDown() {
+ if (store != null) {
+ store.close();
+ }
+ }
+
+ @Test
+ public void shouldRoundTripValueTimestampAndHeadersViaPutAndGet() {
+ final Headers headers = headersWith("schema-id", "42");
+ store.put("k1", ValueTimestampHeaders.make("v1", 1000L, headers));
+
+ final ValueTimestampHeaders<String> result = store.get("k1");
+ assertEquals("v1", result.value());
+ assertEquals(1000L, result.timestamp());
+ assertEquals(headers, result.headers());
+ }
+
+ @Test
+ public void shouldReturnNullForMissingKey() {
+ assertNull(store.get("missing"));
+ }
+
+ @Test
+ public void shouldTreatNullValueAsTombstone() {
+ final Headers headers = headersWith("h", "v");
+ store.put("k", ValueTimestampHeaders.make("value", 10L, headers));
+ assertEquals("value", store.get("k").value());
+
+ store.put("k", null);
+ assertNull(store.get("k"));
+ }
+
+ @Test
+ public void shouldDeleteKeyAndReturnPriorValue() {
+ final Headers headers = headersWith("h", "v");
+ store.put("k", ValueTimestampHeaders.make("value", 10L, headers));
+
+ final ValueTimestampHeaders<String> deleted = store.delete("k");
+ assertEquals("value", deleted.value());
+ assertEquals(10L, deleted.timestamp());
+ assertEquals(headers, deleted.headers());
+ assertNull(store.get("k"));
+ }
+
+ @Test
+ public void shouldPreserveHeadersAcrossRange() {
+ final Headers h1 = headersWith("id", "1");
+ final Headers h2 = headersWith("id", "2");
+ final Headers h3 = headersWith("id", "3");
+
+ store.put("a", ValueTimestampHeaders.make("va", 100L, h1));
+ store.put("b", ValueTimestampHeaders.make("vb", 200L, h2));
+ store.put("c", ValueTimestampHeaders.make("vc", 300L, h3));
+
+ final List<KeyValue<String, ValueTimestampHeaders<String>>> collected
= new ArrayList<>();
+ try (KeyValueIterator<String, ValueTimestampHeaders<String>> it =
store.range("a", "c")) {
+ while (it.hasNext()) {
+ collected.add(it.next());
+ }
+ }
+
+ assertEquals(3, collected.size());
+ assertEquals("a", collected.get(0).key);
+ assertEquals(h1, collected.get(0).value.headers());
+ assertEquals("b", collected.get(1).key);
+ assertEquals(h2, collected.get(1).value.headers());
+ assertEquals("c", collected.get(2).key);
+ assertEquals(h3, collected.get(2).value.headers());
+ }
+
+ @Test
+ public void shouldPreserveHeadersAcrossReverseRange() {
+ final Headers h1 = headersWith("id", "1");
+ final Headers h2 = headersWith("id", "2");
+
+ store.put("a", ValueTimestampHeaders.make("va", 100L, h1));
+ store.put("b", ValueTimestampHeaders.make("vb", 200L, h2));
+
+ final List<KeyValue<String, ValueTimestampHeaders<String>>> collected
= new ArrayList<>();
+ try (KeyValueIterator<String, ValueTimestampHeaders<String>> it =
store.reverseRange("a", "b")) {
+ while (it.hasNext()) {
+ collected.add(it.next());
+ }
+ }
+
+ assertEquals(2, collected.size());
+ assertEquals("b", collected.get(0).key);
+ assertEquals(h2, collected.get(0).value.headers());
+ assertEquals("a", collected.get(1).key);
+ assertEquals(h1, collected.get(1).value.headers());
+ }
+
+ @Test
+ public void shouldReturnEmptyIteratorWhenStoreIsEmpty() {
+ try (KeyValueIterator<String, ValueTimestampHeaders<String>> it =
store.all()) {
+ assertFalse(it.hasNext());
+ }
+ }
+
+ @Test
+ public void shouldPutIfAbsentAndNotOverwriteExisting() {
+ final Headers first = headersWith("h", "first");
+ final Headers second = headersWith("h", "second");
+
+ assertNull(store.putIfAbsent("k", ValueTimestampHeaders.make("v1", 1L,
first)));
+
+ final ValueTimestampHeaders<String> previous =
+ store.putIfAbsent("k", ValueTimestampHeaders.make("v2", 2L,
second));
+ assertEquals("v1", previous.value());
+ assertEquals(first, previous.headers());
+ assertEquals("v1", store.get("k").value());
+ assertEquals(first, store.get("k").headers());
+ }
+
+ @Test
+ public void shouldPreserveEmptyHeaders() {
+ store.put("k", ValueTimestampHeaders.make("v", 10L, new
RecordHeaders()));
+
+ final ValueTimestampHeaders<String> result = store.get("k");
+ assertEquals("v", result.value());
+ assertEquals(new RecordHeaders(), result.headers());
+ assertEquals(0, result.headers().toArray().length);
+ }
+
+ private static Headers headersWith(final String key, final String value) {
+ final Headers headers = new RecordHeaders();
+ headers.add(new RecordHeader(key, value.getBytes()));
+ return headers;
+ }
+}
diff --git
a/streams/src/test/java/org/apache/kafka/streams/state/TimestampedWindowStoreWithHeadersContractTest.java
b/streams/src/test/java/org/apache/kafka/streams/state/TimestampedWindowStoreWithHeadersContractTest.java
new file mode 100644
index 00000000000..ed50ef52648
--- /dev/null
+++
b/streams/src/test/java/org/apache/kafka/streams/state/TimestampedWindowStoreWithHeadersContractTest.java
@@ -0,0 +1,225 @@
+/*
+ * 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.state;
+
+import org.apache.kafka.common.header.Headers;
+import org.apache.kafka.common.header.internals.RecordHeader;
+import org.apache.kafka.common.header.internals.RecordHeaders;
+import org.apache.kafka.common.serialization.Serdes;
+import org.apache.kafka.streams.KeyValue;
+import org.apache.kafka.streams.StreamsConfig;
+import org.apache.kafka.streams.kstream.Windowed;
+import org.apache.kafka.test.InternalMockProcessorContext;
+import org.apache.kafka.test.StreamsTestUtils;
+import org.apache.kafka.test.TestUtils;
+
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+import java.io.File;
+import java.time.Duration;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Properties;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+/**
+ * Contract tests for {@link TimestampedWindowStoreWithHeaders}.
+ */
+public class TimestampedWindowStoreWithHeadersContractTest {
+
+ private static final long WINDOW_SIZE = 10L;
+ private static final long RETENTION = 1000L;
+
+ private TimestampedWindowStoreWithHeaders<String, String> store;
+ private InternalMockProcessorContext<String, String> context;
+
+ @BeforeEach
+ public void setUp() {
+ final File dir = TestUtils.tempDirectory();
+ final Properties props = StreamsTestUtils.getStreamsConfig();
+ context = new InternalMockProcessorContext<>(
+ dir,
+ Serdes.String(),
+ Serdes.String(),
+ new StreamsConfig(props)
+ );
+ store = Stores.timestampedWindowStoreWithHeadersBuilder(
+ Stores.inMemoryWindowStore(
+ "contract-window-store",
+ Duration.ofMillis(RETENTION),
+ Duration.ofMillis(WINDOW_SIZE),
+ false
+ ),
+ Serdes.String(),
+ Serdes.String()
+ ).withLoggingDisabled().withCachingDisabled().build();
+ store.init(context, store);
+ }
+
+ @AfterEach
+ public void tearDown() {
+ if (store != null) {
+ store.close();
+ }
+ }
+
+ @Test
+ public void shouldRoundTripValueTimestampAndHeadersViaPutAndFetch() {
+ final Headers headers = headersWith("schema-id", "42");
+ store.put("k", ValueTimestampHeaders.make("v", 100L, headers), 100L);
+
+ try (WindowStoreIterator<ValueTimestampHeaders<String>> it =
store.fetch("k", 100L, 100L)) {
+ assertTrue(it.hasNext());
+ final KeyValue<Long, ValueTimestampHeaders<String>> next =
it.next();
+ assertEquals(100L, next.key.longValue());
+ assertEquals("v", next.value.value());
+ assertEquals(100L, next.value.timestamp());
+ assertEquals(headers, next.value.headers());
+ assertFalse(it.hasNext());
+ }
+ }
+
+ @Test
+ public void shouldReturnEmptyIteratorForMissingKey() {
+ try (WindowStoreIterator<ValueTimestampHeaders<String>> it =
store.fetch("missing", 0L, 1000L)) {
+ assertFalse(it.hasNext());
+ }
+ }
+
+ @Test
+ public void shouldPreserveHeadersAcrossMultipleWindows() {
+ final Headers h1 = headersWith("id", "w1");
+ final Headers h2 = headersWith("id", "w2");
+ final Headers h3 = headersWith("id", "w3");
+
+ store.put("k", ValueTimestampHeaders.make("v1", 100L, h1), 100L);
+ store.put("k", ValueTimestampHeaders.make("v2", 200L, h2), 200L);
+ store.put("k", ValueTimestampHeaders.make("v3", 300L, h3), 300L);
+
+ final List<KeyValue<Long, Headers>> collected = new ArrayList<>();
+ try (WindowStoreIterator<ValueTimestampHeaders<String>> it =
store.fetch("k", 100L, 300L)) {
+ while (it.hasNext()) {
+ final KeyValue<Long, ValueTimestampHeaders<String>> next =
it.next();
+ collected.add(KeyValue.pair(next.key, next.value.headers()));
+ }
+ }
+
+ assertEquals(3, collected.size());
+ assertEquals(100L, collected.get(0).key.longValue());
+ assertEquals(h1, collected.get(0).value);
+ assertEquals(200L, collected.get(1).key.longValue());
+ assertEquals(h2, collected.get(1).value);
+ assertEquals(300L, collected.get(2).key.longValue());
+ assertEquals(h3, collected.get(2).value);
+ }
+
+ @Test
+ public void shouldPreserveHeadersAcrossBackwardFetch() {
+ final Headers h1 = headersWith("id", "early");
+ final Headers h2 = headersWith("id", "late");
+
+ store.put("k", ValueTimestampHeaders.make("v1", 100L, h1), 100L);
+ store.put("k", ValueTimestampHeaders.make("v2", 200L, h2), 200L);
+
+ final List<KeyValue<Long, Headers>> collected = new ArrayList<>();
+ try (WindowStoreIterator<ValueTimestampHeaders<String>> it =
store.backwardFetch("k", 100L, 200L)) {
+ while (it.hasNext()) {
+ final KeyValue<Long, ValueTimestampHeaders<String>> next =
it.next();
+ collected.add(KeyValue.pair(next.key, next.value.headers()));
+ }
+ }
+
+ assertEquals(2, collected.size());
+ assertEquals(200L, collected.get(0).key.longValue());
+ assertEquals(h2, collected.get(0).value);
+ assertEquals(100L, collected.get(1).key.longValue());
+ assertEquals(h1, collected.get(1).value);
+ }
+
+ @Test
+ public void shouldPreserveHeadersAcrossFetchAll() {
+ final Headers h1 = headersWith("id", "a");
+ final Headers h2 = headersWith("id", "b");
+
+ store.put("a", ValueTimestampHeaders.make("va", 100L, h1), 100L);
+ store.put("b", ValueTimestampHeaders.make("vb", 100L, h2), 100L);
+
+ int count = 0;
+ try (KeyValueIterator<Windowed<String>, ValueTimestampHeaders<String>>
it =
+ store.fetchAll(100L, 100L)) {
+ while (it.hasNext()) {
+ final KeyValue<Windowed<String>,
ValueTimestampHeaders<String>> next = it.next();
+ if ("a".equals(next.key.key())) {
+ assertEquals(h1, next.value.headers());
+ } else if ("b".equals(next.key.key())) {
+ assertEquals(h2, next.value.headers());
+ }
+ count++;
+ }
+ }
+ assertEquals(2, count);
+ }
+
+ @Test
+ public void shouldPreserveHeadersAcrossBackwardFetchAll() {
+ final Headers h1 = headersWith("id", "early");
+ final Headers h2 = headersWith("id", "late");
+
+ store.put("k", ValueTimestampHeaders.make("v1", 100L, h1), 100L);
+ store.put("k", ValueTimestampHeaders.make("v2", 200L, h2), 200L);
+
+ final List<Long> timestamps = new ArrayList<>();
+ try (KeyValueIterator<Windowed<String>, ValueTimestampHeaders<String>>
it =
+ store.backwardFetchAll(100L, 200L)) {
+ while (it.hasNext()) {
+ final KeyValue<Windowed<String>,
ValueTimestampHeaders<String>> next = it.next();
+ timestamps.add(next.key.window().start());
+ if (next.key.window().start() == 100L) {
+ assertEquals(h1, next.value.headers());
+ } else {
+ assertEquals(h2, next.value.headers());
+ }
+ }
+ }
+ assertEquals(2, timestamps.size());
+ assertTrue(timestamps.get(0) >= timestamps.get(1),
+ "backwardFetchAll should return newer windows first");
+ }
+
+ @Test
+ public void shouldPreserveEmptyHeaders() {
+ store.put("k", ValueTimestampHeaders.make("v", 100L, new
RecordHeaders()), 100L);
+
+ try (WindowStoreIterator<ValueTimestampHeaders<String>> it =
store.fetch("k", 100L, 100L)) {
+ assertTrue(it.hasNext());
+ final ValueTimestampHeaders<String> result = it.next().value;
+ assertEquals("v", result.value());
+ assertEquals(new RecordHeaders(), result.headers());
+ }
+ }
+
+ private static Headers headersWith(final String key, final String value) {
+ final Headers headers = new RecordHeaders();
+ headers.add(new RecordHeader(key, value.getBytes()));
+ return headers;
+ }
+}
diff --git
a/streams/src/test/java/org/apache/kafka/streams/state/internals/PlainToHeadersWindowStoreIteratorAdapterTest.java
b/streams/src/test/java/org/apache/kafka/streams/state/internals/PlainToHeadersWindowStoreIteratorAdapterTest.java
new file mode 100644
index 00000000000..fa52269c8d3
--- /dev/null
+++
b/streams/src/test/java/org/apache/kafka/streams/state/internals/PlainToHeadersWindowStoreIteratorAdapterTest.java
@@ -0,0 +1,114 @@
+/*
+ * 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.state.internals;
+
+import org.apache.kafka.streams.KeyValue;
+import org.apache.kafka.streams.state.KeyValueIterator;
+import org.apache.kafka.streams.state.WindowStoreIterator;
+
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.ExtendWith;
+import org.mockito.Mock;
+import org.mockito.junit.jupiter.MockitoExtension;
+import org.mockito.junit.jupiter.MockitoSettings;
+import org.mockito.quality.Strictness;
+
+import static
org.apache.kafka.streams.state.HeadersBytesStore.convertFromPlainToHeaderFormat;
+import static org.junit.jupiter.api.Assertions.assertArrayEquals;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertInstanceOf;
+import static org.junit.jupiter.api.Assertions.assertNull;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+@ExtendWith(MockitoExtension.class)
+@MockitoSettings(strictness = Strictness.STRICT_STUBS)
+public class PlainToHeadersWindowStoreIteratorAdapterTest {
+
+ private static final byte[] PLAIN_VALUE = "value".getBytes();
+ private static final byte[] VALUE_WITH_EMPTY_HEADERS_AND_TS =
+ convertFromPlainToHeaderFormat(PLAIN_VALUE);
+
+ @Mock
+ private WindowStoreIterator<byte[]> innerIterator;
+
+ @Test
+ public void shouldImplementWindowStoreIteratorInterface() {
+ final PlainToHeadersWindowStoreIteratorAdapter adapter =
+ new PlainToHeadersWindowStoreIteratorAdapter(innerIterator);
+
+ assertInstanceOf(WindowStoreIterator.class, adapter);
+ assertInstanceOf(KeyValueIterator.class, adapter);
+ assertInstanceOf(PlainToHeadersIteratorAdapter.class, adapter);
+ }
+
+ @Test
+ public void shouldPrependEmptyHeadersAndSentinelTimestampOnNext() {
+ when(innerIterator.hasNext()).thenReturn(true);
+ when(innerIterator.next()).thenReturn(KeyValue.pair(42L, PLAIN_VALUE));
+
+ final PlainToHeadersWindowStoreIteratorAdapter adapter =
+ new PlainToHeadersWindowStoreIteratorAdapter(innerIterator);
+
+ assertTrue(adapter.hasNext());
+ final KeyValue<Long, byte[]> result = adapter.next();
+ assertEquals(42L, result.key.longValue());
+ assertArrayEquals(VALUE_WITH_EMPTY_HEADERS_AND_TS, result.value);
+ }
+
+ @Test
+ public void shouldReturnNullValueWhenInnerValueIsNull() {
+ when(innerIterator.next()).thenReturn(KeyValue.pair(42L, null));
+
+ final PlainToHeadersWindowStoreIteratorAdapter adapter =
+ new PlainToHeadersWindowStoreIteratorAdapter(innerIterator);
+
+ final KeyValue<Long, byte[]> result = adapter.next();
+ assertEquals(42L, result.key.longValue());
+ assertNull(result.value);
+ }
+
+ @Test
+ public void shouldReturnNullWhenInnerKeyValueIsNull() {
+ when(innerIterator.next()).thenReturn(null);
+
+ final PlainToHeadersWindowStoreIteratorAdapter adapter =
+ new PlainToHeadersWindowStoreIteratorAdapter(innerIterator);
+
+ assertNull(adapter.next());
+ }
+
+ @Test
+ public void shouldDelegatePeekNextKey() {
+ when(innerIterator.peekNextKey()).thenReturn(100L);
+
+ final PlainToHeadersWindowStoreIteratorAdapter adapter =
+ new PlainToHeadersWindowStoreIteratorAdapter(innerIterator);
+
+ assertEquals(100L, adapter.peekNextKey().longValue());
+ }
+
+ @Test
+ public void shouldDelegateClose() {
+ final PlainToHeadersWindowStoreIteratorAdapter adapter =
+ new PlainToHeadersWindowStoreIteratorAdapter(innerIterator);
+
+ adapter.close();
+ verify(innerIterator).close();
+ }
+}
diff --git
a/streams/src/test/java/org/apache/kafka/streams/state/internals/RocksDBMigratingWindowStoreWithHeadersTest.java
b/streams/src/test/java/org/apache/kafka/streams/state/internals/RocksDBMigratingWindowStoreWithHeadersTest.java
new file mode 100644
index 00000000000..353c43a5c29
--- /dev/null
+++
b/streams/src/test/java/org/apache/kafka/streams/state/internals/RocksDBMigratingWindowStoreWithHeadersTest.java
@@ -0,0 +1,190 @@
+/*
+ * 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.state.internals;
+
+import org.apache.kafka.common.utils.Bytes;
+import org.apache.kafka.common.utils.LogCaptureAppender;
+import org.apache.kafka.streams.KeyValue;
+import org.apache.kafka.streams.state.KeyValueIterator;
+import org.apache.kafka.streams.state.internals.metrics.RocksDBMetricsRecorder;
+
+import org.junit.jupiter.api.Test;
+import org.rocksdb.ColumnFamilyDescriptor;
+import org.rocksdb.ColumnFamilyHandle;
+import org.rocksdb.ColumnFamilyOptions;
+import org.rocksdb.DBOptions;
+import org.rocksdb.RocksDB;
+
+import java.io.File;
+import java.util.ArrayList;
+import java.util.List;
+
+import static java.util.Arrays.asList;
+import static org.junit.jupiter.api.Assertions.assertArrayEquals;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertNull;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+/**
+ * Tests for {@link RocksDBMigratingWindowStoreWithHeaders}.
+ */
+public class RocksDBMigratingWindowStoreWithHeadersTest extends
RocksDBStoreTest {
+
+ private final byte[] windowStoreHeadersColumnFamilyName =
+
RocksDBMigratingWindowStoreWithHeaders.WINDOW_STORE_HEADERS_VALUES_COLUMN_FAMILY_NAME;
+
+ RocksDBStore getRocksDBStore() {
+ return new RocksDBMigratingWindowStoreWithHeaders(
+ DB_NAME,
+ "rocksdb",
+ new RocksDBMetricsRecorder(METRICS_SCOPE, DB_NAME));
+ }
+
+ @Test
+ public void shouldOpenNewStoreInRegularMode() {
+ try (final LogCaptureAppender appender =
+
LogCaptureAppender.createAndRegister(RocksDBMigratingWindowStoreWithHeaders.class))
{
+ rocksDBStore.init(context, rocksDBStore);
+
+ assertTrue(appender.getMessages().stream().anyMatch(m ->
m.contains("in regular headers-aware mode")),
+ "Expected log message about regular headers-aware mode, got: "
+ appender.getMessages());
+ }
+
+ try (final KeyValueIterator<Bytes, byte[]> iterator =
rocksDBStore.all()) {
+ assertFalse(iterator.hasNext());
+ }
+ }
+
+ @Test
+ public void shouldOpenExistingHeadersAwareStoreInRegularMode() throws
Exception {
+ final Bytes key = new Bytes("win-key".getBytes());
+ final byte[] value = new byte[] {0x00, 'v', 'a', 'l'};
+
+ rocksDBStore.init(context, rocksDBStore);
+ rocksDBStore.put(key, value);
+ rocksDBStore.close();
+
+ try (final LogCaptureAppender appender =
+
LogCaptureAppender.createAndRegister(RocksDBMigratingWindowStoreWithHeaders.class))
{
+ rocksDBStore.init(context, rocksDBStore);
+
+ assertTrue(appender.getMessages().stream().anyMatch(m ->
m.contains("in regular headers-aware mode")),
+ "Expected regular mode on re-open, got: " +
appender.getMessages());
+ } finally {
+ rocksDBStore.close();
+ }
+
+ verifyValueLandedInHeadersColumnFamily(key, value.length);
+ }
+
+ @Test
+ public void shouldMigrateFromDefaultColumnFamilyWhenLegacyDataExists()
throws Exception {
+ seedDefaultColumnFamilyWithLegacyData();
+
+ try (final LogCaptureAppender appender =
+
LogCaptureAppender.createAndRegister(RocksDBMigratingWindowStoreWithHeaders.class))
{
+ rocksDBStore.init(context, rocksDBStore);
+
+ assertTrue(appender.getMessages().stream().anyMatch(m ->
m.contains("in upgrade mode from plain value format")),
+ "Expected upgrade-mode log, got: " + appender.getMessages());
+ }
+
+ final byte[] legacyKey = "legacy".getBytes();
+ final byte[] migrated = rocksDBStore.get(new Bytes(legacyKey));
+ assertEquals(0x00, migrated[0], "Migrated value must begin with
empty-headers prefix");
+ final byte[] payload = new byte[migrated.length - 1];
+ System.arraycopy(migrated, 1, payload, 0, payload.length);
+ assertArrayEquals("v1".getBytes(), payload);
+
+ assertNull(rocksDBStore.get(new Bytes("unknown".getBytes())));
+
+ rocksDBStore.close();
+ }
+
+ private void seedDefaultColumnFamilyWithLegacyData() {
+ final RocksDBStore plainStore = new RocksDBStore(DB_NAME,
METRICS_SCOPE);
+ try {
+ plainStore.init(context, plainStore);
+ plainStore.put(new Bytes("legacy".getBytes()), "v1".getBytes());
+ } finally {
+ plainStore.close();
+ }
+ }
+
+ private void verifyValueLandedInHeadersColumnFamily(final Bytes key, final
int expectedValueLength) throws Exception {
+ final DBOptions dbOptions = new DBOptions();
+ final ColumnFamilyOptions columnFamilyOptions = new
ColumnFamilyOptions();
+
+ final List<ColumnFamilyDescriptor> columnFamilyDescriptors = asList(
+ new ColumnFamilyDescriptor(RocksDB.DEFAULT_COLUMN_FAMILY,
columnFamilyOptions),
+ new ColumnFamilyDescriptor(windowStoreHeadersColumnFamilyName,
columnFamilyOptions),
+ new
ColumnFamilyDescriptor(RocksDBStore.OFFSETS_COLUMN_FAMILY_NAME,
columnFamilyOptions));
+ final List<ColumnFamilyHandle> columnFamilies = new
ArrayList<>(columnFamilyDescriptors.size());
+
+ RocksDB db = null;
+ ColumnFamilyHandle defaultCf = null;
+ ColumnFamilyHandle headersCf = null;
+ ColumnFamilyHandle offsetsCf = null;
+ try {
+ db = RocksDB.open(
+ dbOptions,
+ new File(new File(context.stateDir(), "rocksdb"),
DB_NAME).getAbsolutePath(),
+ columnFamilyDescriptors,
+ columnFamilies);
+
+ defaultCf = columnFamilies.get(0);
+ headersCf = columnFamilies.get(1);
+ offsetsCf = columnFamilies.get(2);
+
+ assertNull(db.get(defaultCf, key.get()), "DEFAULT CF should not
contain the key");
+ final byte[] inHeadersCf = db.get(headersCf, key.get());
+ assertEquals(expectedValueLength, inHeadersCf.length,
+ "Value should be stored in headers CF with the original
length");
+ } finally {
+ if (offsetsCf != null) {
+ offsetsCf.close();
+ }
+ if (defaultCf != null) {
+ defaultCf.close();
+ }
+ if (headersCf != null) {
+ headersCf.close();
+ }
+ if (db != null) {
+ db.close();
+ }
+ dbOptions.close();
+ columnFamilyOptions.close();
+ }
+ }
+
+ @Test
+ public void shouldIterateOverBothColumnFamiliesInUpgradeMode() {
+ seedDefaultColumnFamilyWithLegacyData();
+ rocksDBStore.init(context, rocksDBStore);
+
+ try (final KeyValueIterator<Bytes, byte[]> it = rocksDBStore.all()) {
+ assertTrue(it.hasNext(), "Iterator should find the legacy key via
on-the-fly conversion");
+ final KeyValue<Bytes, byte[]> kv = it.next();
+ assertArrayEquals("legacy".getBytes(), kv.key.get());
+ assertEquals(0x00, kv.value[0], "Legacy value must be returned
with empty-headers prefix");
+ }
+
+ rocksDBStore.close();
+ }
+}
diff --git
a/streams/src/test/java/org/apache/kafka/streams/state/internals/TimestampedToHeadersStoreAdapterTest.java
b/streams/src/test/java/org/apache/kafka/streams/state/internals/TimestampedToHeadersStoreAdapterTest.java
new file mode 100644
index 00000000000..3b2787d4d6d
--- /dev/null
+++
b/streams/src/test/java/org/apache/kafka/streams/state/internals/TimestampedToHeadersStoreAdapterTest.java
@@ -0,0 +1,390 @@
+/*
+ * 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.state.internals;
+
+import org.apache.kafka.common.utils.Bytes;
+import org.apache.kafka.streams.KeyValue;
+import org.apache.kafka.streams.query.KeyQuery;
+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.query.RangeQuery;
+import org.apache.kafka.streams.state.KeyValueIterator;
+import org.apache.kafka.streams.state.KeyValueStore;
+import org.apache.kafka.streams.state.TimestampedBytesStore;
+
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.ExtendWith;
+import org.mockito.Mock;
+import org.mockito.junit.jupiter.MockitoExtension;
+import org.mockito.junit.jupiter.MockitoSettings;
+import org.mockito.quality.Strictness;
+
+import java.util.Arrays;
+
+import static
org.apache.kafka.streams.state.HeadersBytesStore.convertToHeaderFormat;
+import static org.junit.jupiter.api.Assertions.assertArrayEquals;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertInstanceOf;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertNull;
+import static org.junit.jupiter.api.Assertions.assertSame;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.eq;
+import static org.mockito.Mockito.lenient;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+import static org.mockito.Mockito.withSettings;
+
+@ExtendWith(MockitoExtension.class)
+@MockitoSettings(strictness = Strictness.STRICT_STUBS)
+public class TimestampedToHeadersStoreAdapterTest {
+
+ @Mock
+ private KeyValueIterator<Bytes, byte[]> mockIterator;
+
+ @SuppressWarnings("unchecked")
+ private KeyValueStore<Bytes, byte[]> mockStore;
+
+ private TimestampedToHeadersStoreAdapter adapter;
+
+ @SuppressWarnings("unchecked")
+ @BeforeEach
+ public void setUp() {
+ mockStore = mock(KeyValueStore.class,
withSettings().extraInterfaces(TimestampedBytesStore.class));
+ // lenient: this fixture stub is consumed by the adapter constructor
for most tests, but the
+ // constructor-validation tests build their own store and never touch
this one.
+ lenient().when(mockStore.persistent()).thenReturn(true);
+ adapter = new TimestampedToHeadersStoreAdapter(mockStore);
+ }
+
+ @Test
+ @SuppressWarnings("unchecked")
+ public void shouldThrowIfStoreIsNotPersistent() {
+ final KeyValueStore<Bytes, byte[]> nonPersistentStore =
+ mock(KeyValueStore.class,
withSettings().extraInterfaces(TimestampedBytesStore.class));
+ when(nonPersistentStore.persistent()).thenReturn(false);
+
+ final IllegalArgumentException exception = assertThrows(
+ IllegalArgumentException.class,
+ () -> new TimestampedToHeadersStoreAdapter(nonPersistentStore)
+ );
+
+ assertTrue(exception.getMessage().contains("Provided store must be a
persistent store"));
+ }
+
+ @Test
+ @SuppressWarnings("unchecked")
+ public void shouldThrowIfStoreIsNotTimestamped() {
+ final KeyValueStore<Bytes, byte[]> nonTimestampedStore =
mock(KeyValueStore.class);
+ when(nonTimestampedStore.persistent()).thenReturn(true);
+
+ final IllegalArgumentException exception = assertThrows(
+ IllegalArgumentException.class,
+ () -> new TimestampedToHeadersStoreAdapter(nonTimestampedStore)
+ );
+
+ assertTrue(exception.getMessage().contains("Provided store must be a
timestamped store"));
+ }
+
+ @Test
+ public void shouldPutRawTimestampedValueToStore() {
+ final Bytes key = new Bytes("key".getBytes());
+ final byte[] rawTimestampedValue =
+ new byte[] {0, 0, 0, 0, 0, 0, 0, 42, 'v', 'a', 'l'};
+ final byte[] valueWithHeaders =
convertToHeaderFormat(rawTimestampedValue);
+
+ adapter.put(key, valueWithHeaders);
+
+ verify(mockStore).put(eq(key), eq(rawTimestampedValue));
+ }
+
+ @Test
+ public void shouldGetAndConvertToHeaderFormat() {
+ final Bytes key = new Bytes("key".getBytes());
+ final byte[] rawTimestampedValue =
+ new byte[] {0, 0, 0, 0, 0, 0, 0, 42, 'v', 'a', 'l'};
+ when(mockStore.get(key)).thenReturn(rawTimestampedValue);
+
+ final byte[] result = adapter.get(key);
+
+ assertArrayEquals(convertToHeaderFormat(rawTimestampedValue), result);
+ }
+
+ @Test
+ public void shouldReturnNullWhenStoreReturnsNull() {
+ final Bytes key = new Bytes("key".getBytes());
+ when(mockStore.get(key)).thenReturn(null);
+
+ assertNull(adapter.get(key));
+ }
+
+ @Test
+ public void shouldPutIfAbsentAndConvertResult() {
+ final Bytes key = new Bytes("key".getBytes());
+ final byte[] rawTimestampedValue =
+ new byte[] {0, 0, 0, 0, 0, 0, 0, 42, 'v', 'a', 'l'};
+ final byte[] valueWithHeaders =
convertToHeaderFormat(rawTimestampedValue);
+ final byte[] oldRawValue =
+ new byte[] {0, 0, 0, 0, 0, 0, 0, 10, 'o', 'l', 'd'};
+ when(mockStore.putIfAbsent(eq(key),
eq(rawTimestampedValue))).thenReturn(oldRawValue);
+
+ final byte[] result = adapter.putIfAbsent(key, valueWithHeaders);
+
+ assertArrayEquals(convertToHeaderFormat(oldRawValue), result);
+ }
+
+ @Test
+ public void shouldDeleteAndConvertResult() {
+ final Bytes key = new Bytes("key".getBytes());
+ final byte[] oldRawValue =
+ new byte[] {0, 0, 0, 0, 0, 0, 0, 10, 'o', 'l', 'd'};
+ when(mockStore.delete(key)).thenReturn(oldRawValue);
+
+ final byte[] result = adapter.delete(key);
+
+ assertArrayEquals(convertToHeaderFormat(oldRawValue), result);
+ }
+
+ @Test
+ public void shouldPutAllEntries() {
+ final Bytes key1 = new Bytes("key1".getBytes());
+ final Bytes key2 = new Bytes("key2".getBytes());
+ final byte[] rawValue1 =
+ new byte[] {0, 0, 0, 0, 0, 0, 0, 1, 'v', '1'};
+ final byte[] rawValue2 =
+ new byte[] {0, 0, 0, 0, 0, 0, 0, 2, 'v', '2'};
+ final byte[] value1 = convertToHeaderFormat(rawValue1);
+ final byte[] value2 = convertToHeaderFormat(rawValue2);
+
+ adapter.putAll(Arrays.asList(
+ KeyValue.pair(key1, value1),
+ KeyValue.pair(key2, value2)
+ ));
+
+ verify(mockStore).put(eq(key1), eq(rawValue1));
+ verify(mockStore).put(eq(key2), eq(rawValue2));
+ }
+
+ @Test
+ public void shouldWrapRangeIterator() {
+ final Bytes from = new Bytes("a".getBytes());
+ final Bytes to = new Bytes("z".getBytes());
+ when(mockStore.range(from, to)).thenReturn(mockIterator);
+
+ final KeyValueIterator<Bytes, byte[]> result = adapter.range(from, to);
+
+ assertNotNull(result);
+ assertInstanceOf(TimestampedToHeadersIteratorAdapter.class, result);
+ }
+
+ @Test
+ public void shouldWrapReverseRangeIterator() {
+ final Bytes from = new Bytes("a".getBytes());
+ final Bytes to = new Bytes("z".getBytes());
+ when(mockStore.reverseRange(from, to)).thenReturn(mockIterator);
+
+ final KeyValueIterator<Bytes, byte[]> result =
adapter.reverseRange(from, to);
+
+ assertNotNull(result);
+ assertInstanceOf(TimestampedToHeadersIteratorAdapter.class, result);
+ }
+
+ @Test
+ public void shouldWrapAllIterator() {
+ when(mockStore.all()).thenReturn(mockIterator);
+
+ final KeyValueIterator<Bytes, byte[]> result = adapter.all();
+
+ assertNotNull(result);
+ assertInstanceOf(TimestampedToHeadersIteratorAdapter.class, result);
+ }
+
+ @Test
+ public void shouldWrapReverseAllIterator() {
+ when(mockStore.reverseAll()).thenReturn(mockIterator);
+
+ final KeyValueIterator<Bytes, byte[]> result = adapter.reverseAll();
+
+ assertNotNull(result);
+ assertInstanceOf(TimestampedToHeadersIteratorAdapter.class, result);
+ }
+
+ @Test
+ public void shouldWrapPrefixScanIterator() {
+ when(mockStore.prefixScan(any(), any())).thenReturn(mockIterator);
+
+ final KeyValueIterator<Bytes, byte[]> result =
+ adapter.prefixScan("prefix", (topic, data) -> data.getBytes());
+
+ assertNotNull(result);
+ assertInstanceOf(TimestampedToHeadersIteratorAdapter.class, result);
+ }
+
+ @Test
+ public void shouldHandleKeyQuery() {
+ final Bytes key = new Bytes("test-key".getBytes());
+ final byte[] rawTimestampedValue =
+ new byte[] {0, 0, 0, 0, 0, 0, 0, 42, 'v', 'a', 'l'};
+ final KeyQuery<Bytes, byte[]> query = KeyQuery.withKey(key);
+
+ final QueryResult<byte[]> mockResult =
QueryResult.forResult(rawTimestampedValue);
+ when(mockStore.query(eq(query), any(PositionBound.class),
any(QueryConfig.class)))
+ .thenReturn(mockResult);
+
+ final QueryResult<byte[]> result =
+ adapter.query(query, PositionBound.unbounded(), new
QueryConfig(false));
+
+ assertTrue(result.isSuccess());
+ assertArrayEquals(convertToHeaderFormat(rawTimestampedValue),
result.getResult());
+ }
+
+ @Test
+ public void shouldHandleKeyQueryWithNullResult() {
+ final Bytes key = new Bytes("test-key".getBytes());
+ final KeyQuery<Bytes, byte[]> query = KeyQuery.withKey(key);
+
+ final QueryResult<byte[]> mockResult = QueryResult.forResult(null);
+ when(mockStore.query(eq(query), any(PositionBound.class),
any(QueryConfig.class)))
+ .thenReturn(mockResult);
+
+ final QueryResult<byte[]> result =
+ adapter.query(query, PositionBound.unbounded(), new
QueryConfig(false));
+
+ assertTrue(result.isSuccess());
+ assertNull(result.getResult());
+ }
+
+ @Test
+ public void shouldHandleFailedKeyQuery() {
+ final Bytes key = new Bytes("test-key".getBytes());
+ final KeyQuery<Bytes, byte[]> query = KeyQuery.withKey(key);
+
+ final QueryResult<byte[]> mockResult =
QueryResult.forUnknownQueryType(query, mockStore);
+ when(mockStore.query(eq(query), any(PositionBound.class),
any(QueryConfig.class)))
+ .thenReturn(mockResult);
+
+ final QueryResult<byte[]> result =
+ adapter.query(query, PositionBound.unbounded(), new
QueryConfig(false));
+
+ assertFalse(result.isSuccess());
+ }
+
+ @Test
+ public void shouldHandleRangeQuery() {
+ final RangeQuery<Bytes, byte[]> query = RangeQuery.withRange(
+ new Bytes("a".getBytes()),
+ new Bytes("z".getBytes())
+ );
+
+ final QueryResult<KeyValueIterator<Bytes, byte[]>> mockResult =
+ QueryResult.forResult(mockIterator);
+ when(mockStore.query(eq(query), any(PositionBound.class),
any(QueryConfig.class)))
+ .thenReturn(mockResult);
+
+ final QueryResult<KeyValueIterator<Bytes, byte[]>> result =
adapter.query(
+ query,
+ PositionBound.unbounded(),
+ new QueryConfig(false)
+ );
+
+ assertTrue(result.isSuccess());
+ assertNotNull(result.getResult());
+ assertInstanceOf(TimestampedToHeadersIteratorAdapter.class,
result.getResult());
+ }
+
+ @Test
+ @SuppressWarnings("unchecked")
+ public void shouldDelegateOtherQueryTypesToStore() {
+ // Any query that is neither KeyQuery nor RangeQuery falls through to
the
+ // else branch and is passed straight to the underlying store,
unchanged.
+ final Query<String> query = mock(Query.class);
+ final QueryResult<String> mockResult =
QueryResult.forResult("delegated");
+ when(mockStore.query(eq(query), any(PositionBound.class),
any(QueryConfig.class)))
+ .thenReturn(mockResult);
+
+ final QueryResult<String> result =
+ adapter.query(query, PositionBound.unbounded(), new
QueryConfig(false));
+
+ assertSame(mockResult, result);
+ assertTrue(result.isSuccess());
+ assertEquals("delegated", result.getResult());
+ }
+
+ @Test
+ public void shouldCollectExecutionInfoForKeyQuery() {
+ final Bytes key = new Bytes("test-key".getBytes());
+ final byte[] rawTimestampedValue =
+ new byte[] {0, 0, 0, 0, 0, 0, 0, 42, 'v', 'a', 'l'};
+ final KeyQuery<Bytes, byte[]> query = KeyQuery.withKey(key);
+
+ final QueryResult<byte[]> mockResult =
QueryResult.forResult(rawTimestampedValue);
+ when(mockStore.query(eq(query), any(PositionBound.class),
any(QueryConfig.class)))
+ .thenReturn(mockResult);
+
+ final QueryResult<byte[]> result =
+ adapter.query(query, PositionBound.unbounded(), new
QueryConfig(true));
+
+ assertTrue(result.isSuccess());
+ assertFalse(result.getExecutionInfo().isEmpty(),
+ "Expected execution info to be collected");
+ final String executionInfo = String.join("\n",
result.getExecutionInfo());
+ assertTrue(executionInfo.contains("Handled in"));
+
assertTrue(executionInfo.contains(TimestampedToHeadersStoreAdapter.class.getName()));
+ }
+
+ @Test
+ public void shouldDelegateName() {
+ when(mockStore.name()).thenReturn("test-store");
+
+ assertEquals("test-store", adapter.name());
+ }
+
+ @Test
+ public void shouldReturnTrueForPersistent() {
+ assertTrue(adapter.persistent());
+ }
+
+ @Test
+ public void shouldDelegateIsOpen() {
+ when(mockStore.isOpen()).thenReturn(true);
+
+ assertTrue(adapter.isOpen());
+ }
+
+ @Test
+ public void shouldDelegateApproximateNumEntries() {
+ when(mockStore.approximateNumEntries()).thenReturn(42L);
+
+ assertEquals(42L, adapter.approximateNumEntries());
+ }
+
+ @Test
+ public void shouldDelegateGetPosition() {
+ when(mockStore.getPosition()).thenReturn(null);
+
+ adapter.getPosition();
+
+ verify(mockStore).getPosition();
+ }
+}
