From dc097af38e9198b9c4954c36268f704c4f6c68ae Mon Sep 17 00:00:00 2001 From: Alieh Saeedi Date: Mon, 27 Jul 2026 13:46:00 +0200 Subject: [PATCH 1/6] KAFKA-20413: Add headers-aware RocksDB list-value store [1/2] Adds the dual-column-family RocksDB store that lets an outer-join ListValueStore written in the pre-headers PLAIN format be reopened and read in the HEADERS format (dsl.store.format=headers) without corrupting old data. The store is not wired into any topology yet; the DSL change that enables it follows in PR 2/2. The outer-join store persists, per key, a ListSerde blob whose elements are single serialized values, so it cannot reuse RocksDBTimestampedStoreWithHeaders: that store's whole-value [0x00][ts=-1] converter would corrupt the list encoding. RocksDBListValueStoreWithHeaders instead keeps legacy PLAIN blobs in the DEFAULT column family and lifts each list element to the empty-headers format on read/write (prepend 0x00 per element) via DualColumnFamilyAccessor, migrating them into a new listValueWithHeaders column family. New stores open directly on that column family through a SingleColumnFamilyAccessor. RocksDBStore reports a clear error on an unsupported HEADERS-to-PLAIN downgrade of the new store, mirroring the existing timestamped/headers downgrade guards. Tests cover the blob converter (including the right value that the naive read-through silently truncated and the left value that threw SerializationException) and an end-to-end RocksDB upgrade: write with the pre-headers PLAIN store, reopen as the dual-CF HEADERS store, and assert correct reads for left/right/multi-element values plus appends across the upgrade boundary. ListValueStore is internal, so no KIP is required. Co-Authored-By: Claude Opus 5 (1M context) --- .../internals/ListValueStoreUpgradeUtils.java | 66 ++++++++ ...sDBListValueHeadersBytesStoreSupplier.java | 49 ++++++ .../RocksDBListValueStoreWithHeaders.java | 108 +++++++++++++ .../streams/state/internals/RocksDBStore.java | 7 + .../ListValueStoreUpgradeUtilsTest.java | 139 ++++++++++++++++ .../RocksDBListValueStoreWithHeadersTest.java | 151 ++++++++++++++++++ 6 files changed, 520 insertions(+) create mode 100644 streams/src/main/java/org/apache/kafka/streams/state/internals/ListValueStoreUpgradeUtils.java create mode 100644 streams/src/main/java/org/apache/kafka/streams/state/internals/RocksDBListValueHeadersBytesStoreSupplier.java create mode 100644 streams/src/main/java/org/apache/kafka/streams/state/internals/RocksDBListValueStoreWithHeaders.java create mode 100644 streams/src/test/java/org/apache/kafka/streams/state/internals/ListValueStoreUpgradeUtilsTest.java create mode 100644 streams/src/test/java/org/apache/kafka/streams/state/internals/RocksDBListValueStoreWithHeadersTest.java diff --git a/streams/src/main/java/org/apache/kafka/streams/state/internals/ListValueStoreUpgradeUtils.java b/streams/src/main/java/org/apache/kafka/streams/state/internals/ListValueStoreUpgradeUtils.java new file mode 100644 index 0000000000000..491455ffecb48 --- /dev/null +++ b/streams/src/main/java/org/apache/kafka/streams/state/internals/ListValueStoreUpgradeUtils.java @@ -0,0 +1,66 @@ +/* + * 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.serialization.Serde; +import org.apache.kafka.common.serialization.Serdes; +import org.apache.kafka.streams.state.HeadersBytesStore; + +import java.util.ArrayList; +import java.util.List; + +/** + * Helpers for migrating the outer-join {@link ListValueStore} from the pre-headers PLAIN element + * format to the HEADERS element format (KIP-1271, added for AK 4.4). + *

+ * The store persists, per key, a {@link Serdes#ListSerde} blob whose elements are single serialized + * values. The element encoding differs by {@code dsl.store.format}: + *

+ * A PLAIN element becomes a HEADERS element with empty headers simply by prepending a single + * {@code 0x00} byte (the empty-headers varint) — see {@link HeadersBytesStore#convertToHeaderFormat}. + * So a whole PLAIN list blob is converted by prepending {@code 0x00} to each element and re-serializing + * the same {@code ListSerde}. + */ +final class ListValueStoreUpgradeUtils { + + // Must match ListValueStore.LIST_SERDE. + @SuppressWarnings("unchecked") + private static final Serde> LIST_SERDE = Serdes.ListSerde(ArrayList.class, Serdes.ByteArray()); + + private ListValueStoreUpgradeUtils() {} + + /** + * Converts a whole PLAIN list blob into the HEADERS list blob by lifting each element to the + * empty-headers format. {@code null} (a tombstone / whole-list delete) is passed through. + */ + static byte[] convertPlainListBlobToHeadersListBlob(final byte[] plainListBlob) { + if (plainListBlob == null) { + return null; + } + final List plainElements = LIST_SERDE.deserializer().deserialize(null, plainListBlob); + final List headersElements = new ArrayList<>(plainElements.size()); + for (final byte[] element : plainElements) { + // convertToHeaderFormat(null) returns null, preserving any null list members. + headersElements.add(HeadersBytesStore.convertToHeaderFormat(element)); + } + return LIST_SERDE.serializer().serialize(null, headersElements); + } +} diff --git a/streams/src/main/java/org/apache/kafka/streams/state/internals/RocksDBListValueHeadersBytesStoreSupplier.java b/streams/src/main/java/org/apache/kafka/streams/state/internals/RocksDBListValueHeadersBytesStoreSupplier.java new file mode 100644 index 0000000000000..ba6edad9f8537 --- /dev/null +++ b/streams/src/main/java/org/apache/kafka/streams/state/internals/RocksDBListValueHeadersBytesStoreSupplier.java @@ -0,0 +1,49 @@ +/* + * 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.state.KeyValueBytesStoreSupplier; +import org.apache.kafka.streams.state.KeyValueStore; + +/** + * Supplies the persistent, dual-column-family {@link RocksDBListValueStoreWithHeaders} used as the + * bytes store for the outer-join {@link ListValueStore} in HEADERS mode. + */ +public class RocksDBListValueHeadersBytesStoreSupplier implements KeyValueBytesStoreSupplier { + + private final String name; + + public RocksDBListValueHeadersBytesStoreSupplier(final String name) { + this.name = name; + } + + @Override + public String name() { + return name; + } + + @Override + public KeyValueStore get() { + return new RocksDBListValueStoreWithHeaders(name, metricsScope()); + } + + @Override + public String metricsScope() { + return "rocksdb"; + } +} diff --git a/streams/src/main/java/org/apache/kafka/streams/state/internals/RocksDBListValueStoreWithHeaders.java b/streams/src/main/java/org/apache/kafka/streams/state/internals/RocksDBListValueStoreWithHeaders.java new file mode 100644 index 0000000000000..0ccf9dfc4c4d6 --- /dev/null +++ b/streams/src/main/java/org/apache/kafka/streams/state/internals/RocksDBListValueStoreWithHeaders.java @@ -0,0 +1,108 @@ +/* + * 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.state.internals.metrics.RocksDBMetricsRecorder; + +import org.rocksdb.ColumnFamilyDescriptor; +import org.rocksdb.ColumnFamilyHandle; +import org.rocksdb.ColumnFamilyOptions; +import org.rocksdb.DBOptions; +import org.rocksdb.RocksDB; +import org.rocksdb.RocksIterator; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.nio.charset.StandardCharsets; +import java.util.List; + +/** + * A persistent bytes key-value store for the outer-join {@link ListValueStore} in HEADERS mode. + *

+ * The store keeps two column families so it can be upgraded in place from a pre-headers (PLAIN) store + * without corrupting existing data (KIP-1271 dual-column-family pattern, mirroring + * {@link RocksDBTimestampedStoreWithHeaders}): + *

    + *
  • DEFAULT: legacy PLAIN list blobs written by the pre-headers version;
    • + *
  • {@code listValueWithHeaders}: list blobs whose elements carry inline headers.
    • + *
+ * When the DEFAULT column family holds data at open time, a {@link DualColumnFamilyAccessor} lifts each + * legacy blob to the headers format on read/write via + * {@link ListValueStoreUpgradeUtils#convertPlainListBlobToHeadersListBlob} and migrates it forward. + * Otherwise a {@link RocksDBStore.SingleColumnFamilyAccessor} over the headers column family is used. + *

+ * This class intentionally does NOT implement {@code HeadersBytesStore}; see + * {@link HeadersAwareListValueStore} for why. + */ +public class RocksDBListValueStoreWithHeaders extends RocksDBStore { + + private static final Logger log = LoggerFactory.getLogger(RocksDBListValueStoreWithHeaders.class); + + static final byte[] LIST_VALUE_WITH_HEADERS_CF_NAME = + "listValueWithHeaders".getBytes(StandardCharsets.UTF_8); + + RocksDBListValueStoreWithHeaders(final String name, + final String metricsScope) { + super(name, metricsScope); + } + + RocksDBListValueStoreWithHeaders(final String name, + final String parentDir, + final RocksDBMetricsRecorder metricsRecorder) { + super(name, parentDir, metricsRecorder); + } + + @Override + void openRocksDB(final DBOptions dbOptions, + final ColumnFamilyOptions columnFamilyOptions) { + final List columnFamilies = openRocksDB( + dbOptions, + new ColumnFamilyDescriptor(RocksDB.DEFAULT_COLUMN_FAMILY, columnFamilyOptions), + new ColumnFamilyDescriptor(LIST_VALUE_WITH_HEADERS_CF_NAME, columnFamilyOptions), + new ColumnFamilyDescriptor(OFFSETS_COLUMN_FAMILY_NAME, offsetsCFOptions()) + ); + + final ColumnFamilyHandle defaultCf = columnFamilies.get(0); + final ColumnFamilyHandle listValueWithHeadersCf = columnFamilies.get(1); + final ColumnFamilyHandle offsetsCf = columnFamilies.get(2); + + // If the DEFAULT column family holds data, we are upgrading from a plain list-value store. + try (final RocksIterator defaultIter = db.newIterator(defaultCf)) { + defaultIter.seekToFirst(); + if (defaultIter.isValid()) { + log.info("Opening store {} in upgrade mode from plain list-value store", name); + cfAccessor = new DualColumnFamilyAccessor( + offsetsCf, + defaultCf, + listValueWithHeadersCf, + ListValueStoreUpgradeUtils::convertPlainListBlobToHeadersListBlob, + this, + open + ); + } else { + log.info("Opening store {} in regular headers-aware mode", name); + cfAccessor = new SingleColumnFamilyAccessor(offsetsCf, listValueWithHeadersCf); + defaultCf.close(); + } + } catch (final RuntimeException e) { + for (final ColumnFamilyHandle handle : columnFamilies) { + handle.close(); + } + throw e; + } + } +} diff --git a/streams/src/main/java/org/apache/kafka/streams/state/internals/RocksDBStore.java b/streams/src/main/java/org/apache/kafka/streams/state/internals/RocksDBStore.java index 398ce2f88d04c..06edc126ff209 100644 --- a/streams/src/main/java/org/apache/kafka/streams/state/internals/RocksDBStore.java +++ b/streams/src/main/java/org/apache/kafka/streams/state/internals/RocksDBStore.java @@ -391,6 +391,13 @@ protected List openRocksDB(final DBOptions dbOptions, } } + if (Arrays.equals(existingFamily, RocksDBListValueStoreWithHeaders.LIST_VALUE_WITH_HEADERS_CF_NAME)) { + throw new ProcessorStateException( + "Store " + name + " is a headers-aware list-value store and cannot be opened as a regular store. " + + "Downgrade from headers-aware to regular store is not supported. " + + "To downgrade, you can delete the local state in the state directory, and rebuild the store from the changelog."); + } + final String unexpectedFamily = new String(existingFamily, StandardCharsets.UTF_8); throw new ProcessorStateException( "Unexpected column family '" + unexpectedFamily + "' found in store " + name + ". " + diff --git a/streams/src/test/java/org/apache/kafka/streams/state/internals/ListValueStoreUpgradeUtilsTest.java b/streams/src/test/java/org/apache/kafka/streams/state/internals/ListValueStoreUpgradeUtilsTest.java new file mode 100644 index 0000000000000..4c796d4025a96 --- /dev/null +++ b/streams/src/test/java/org/apache/kafka/streams/state/internals/ListValueStoreUpgradeUtilsTest.java @@ -0,0 +1,139 @@ +/* + * 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.header.Headers; +import org.apache.kafka.common.header.internals.RecordHeaders; +import org.apache.kafka.common.serialization.Serde; +import org.apache.kafka.common.serialization.Serdes; +import org.apache.kafka.streams.state.AggregationWithHeaders; + +import org.junit.jupiter.api.Test; + +import java.util.ArrayList; +import java.util.Arrays; +import java.util.List; + +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; + +public class ListValueStoreUpgradeUtilsTest { + + private static final String TOPIC = "t"; + private static final Headers EMPTY = new RecordHeaders(); + + @SuppressWarnings("unchecked") + private static final Serde> LIST_SERDE = Serdes.ListSerde(ArrayList.class, Serdes.ByteArray()); + + private final LeftOrRightValueSerde plainSerde = + new LeftOrRightValueSerde<>(Serdes.String(), Serdes.String()); + private final AggregationWithHeadersSerde> headersSerde = + new AggregationWithHeadersSerde<>(plainSerde); + + private byte[] plainElement(final LeftOrRightValue value) { + return plainSerde.serializer().serialize(TOPIC, EMPTY, value); + } + + private byte[] plainListBlob(final List> values) { + final List elements = new ArrayList<>(); + for (final LeftOrRightValue v : values) { + elements.add(plainElement(v)); + } + return LIST_SERDE.serializer().serialize(null, elements); + } + + private List>> readHeadersListBlob(final byte[] blob) { + final List elements = LIST_SERDE.deserializer().deserialize(null, blob); + final List>> out = new ArrayList<>(); + for (final byte[] e : elements) { + out.add(headersSerde.deserializer().deserialize(TOPIC, e)); + } + return out; + } + + @Test + public void shouldConvertRightValueWithoutCorruption() { + final byte[] plain = plainListBlob(List.of(LeftOrRightValue.makeRightValue("right"))); + + final byte[] converted = ListValueStoreUpgradeUtils.convertPlainListBlobToHeadersListBlob(plain); + final List>> result = readHeadersListBlob(converted); + + assertEquals(1, result.size()); + // The pre-fix bug silently corrupted right values (right -> ight); assert it's intact now. + assertEquals("right", result.get(0).aggregation().rightValue()); + assertNull(result.get(0).aggregation().leftValue()); + assertFalse(result.get(0).headers().iterator().hasNext(), "lifted headers should be empty"); + } + + @Test + public void shouldConvertLeftValueThatPreviouslyThrew() { + final byte[] plain = plainListBlob(List.of(LeftOrRightValue.makeLeftValue("left"))); + + final byte[] converted = ListValueStoreUpgradeUtils.convertPlainListBlobToHeadersListBlob(plain); + final List>> result = readHeadersListBlob(converted); + + assertEquals(1, result.size()); + assertEquals("left", result.get(0).aggregation().leftValue()); + assertNull(result.get(0).aggregation().rightValue()); + } + + @Test + public void shouldConvertMultiElementMixedList() { + final byte[] plain = plainListBlob(List.of( + LeftOrRightValue.makeLeftValue("a"), + LeftOrRightValue.makeRightValue("b"), + LeftOrRightValue.makeLeftValue("c") + )); + + final List>> result = + readHeadersListBlob(ListValueStoreUpgradeUtils.convertPlainListBlobToHeadersListBlob(plain)); + + assertEquals(3, result.size()); + assertEquals("a", result.get(0).aggregation().leftValue()); + assertEquals("b", result.get(1).aggregation().rightValue()); + assertEquals("c", result.get(2).aggregation().leftValue()); + } + + @Test + public void shouldPassThroughNull() { + assertNull(ListValueStoreUpgradeUtils.convertPlainListBlobToHeadersListBlob(null)); + } + + @Test + public void shouldConvertEmptyListToEmptyList() { + final byte[] emptyBlob = LIST_SERDE.serializer().serialize(null, new ArrayList<>()); + + final byte[] converted = ListValueStoreUpgradeUtils.convertPlainListBlobToHeadersListBlob(emptyBlob); + + assertEquals(0, LIST_SERDE.deserializer().deserialize(null, converted).size()); + } + + @Test + public void convertedElementShouldBePlainElementPrefixedWithZeroByte() { + final byte[] plainEl = plainElement(LeftOrRightValue.makeRightValue("x")); + final byte[] plain = plainListBlob(List.of(LeftOrRightValue.makeRightValue("x"))); + + final List convertedElements = + LIST_SERDE.deserializer().deserialize(null, ListValueStoreUpgradeUtils.convertPlainListBlobToHeadersListBlob(plain)); + + final byte[] expected = new byte[plainEl.length + 1]; + System.arraycopy(plainEl, 0, expected, 1, plainEl.length); // expected[0] == 0x00 + assertTrue(Arrays.equals(expected, convertedElements.get(0))); + } +} diff --git a/streams/src/test/java/org/apache/kafka/streams/state/internals/RocksDBListValueStoreWithHeadersTest.java b/streams/src/test/java/org/apache/kafka/streams/state/internals/RocksDBListValueStoreWithHeadersTest.java new file mode 100644 index 0000000000000..0bde6cc0c84cb --- /dev/null +++ b/streams/src/test/java/org/apache/kafka/streams/state/internals/RocksDBListValueStoreWithHeadersTest.java @@ -0,0 +1,151 @@ +/* + * 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.metrics.Metrics; +import org.apache.kafka.common.serialization.Serdes; +import org.apache.kafka.common.utils.Time; +import org.apache.kafka.common.utils.internals.LogContext; +import org.apache.kafka.streams.KeyValue; +import org.apache.kafka.streams.processor.internals.MockStreamsMetrics; +import org.apache.kafka.streams.state.AggregationWithHeaders; +import org.apache.kafka.streams.state.KeyValueStore; +import org.apache.kafka.streams.state.Stores; +import org.apache.kafka.test.InternalMockProcessorContext; +import org.apache.kafka.test.MockRecordCollector; +import org.apache.kafka.test.TestUtils; + +import org.junit.jupiter.api.Test; + +import java.io.File; +import java.util.List; + +import static org.apache.kafka.test.StreamsTestUtils.toListAndCloseIterator; +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertNull; + +/** + * Verifies that a persistent outer-join list-value store written by a pre-headers (PLAIN) version can + * be reopened as a HEADERS-format {@link RocksDBListValueStoreWithHeaders} and read back correctly via + * the dual-column-family migration — without the silent corruption / {@code SerializationException} + * that a plain read-through {@code AggregationWithHeadersSerde} would produce. + */ +public class RocksDBListValueStoreWithHeadersTest { + + private static final String STORE_NAME = "outer-join-list-store"; + + private final File baseDir = TestUtils.tempDirectory("list-upgrade"); + + private InternalMockProcessorContext newContext() { + final MockRecordCollector recordCollector = new MockRecordCollector(); + final ThreadCache cache = new ThreadCache(new LogContext("test"), 0, new MockStreamsMetrics(new Metrics())); + final InternalMockProcessorContext context = new InternalMockProcessorContext<>( + baseDir, Serdes.String(), null, recordCollector, cache); + context.setTime(1L); + return context; + } + + private KeyValueStore> buildPlainStore() { + return new ListValueStoreBuilder>( + Stores.persistentKeyValueStore(STORE_NAME), + Serdes.String(), + new LeftOrRightValueSerde<>(Serdes.String(), Serdes.String()), + Time.SYSTEM) + .withCachingDisabled() + .withLoggingDisabled() + .build(); + } + + private KeyValueStore>> buildHeadersStore() { + return new ListValueStoreBuilder>>( + new RocksDBListValueHeadersBytesStoreSupplier(STORE_NAME), + Serdes.String(), + new AggregationWithHeadersSerde<>(new LeftOrRightValueSerde<>(Serdes.String(), Serdes.String())), + Time.SYSTEM) + .withCachingDisabled() + .withLoggingDisabled() + .build(); + } + + @Test + public void shouldReadPlainDataAfterUpgradeToHeadersStore() { + // Phase 1 — write with the pre-headers PLAIN list store. + final KeyValueStore> plainStore = buildPlainStore(); + final InternalMockProcessorContext ctx1 = newContext(); + plainStore.init(ctx1, plainStore); + // "right" is the value that got silently truncated pre-fix; "left" is the one that threw. + plainStore.put("k1", LeftOrRightValue.makeRightValue("right")); + plainStore.put("k1", LeftOrRightValue.makeLeftValue("left")); // second element, same key (list) + plainStore.put("k2", LeftOrRightValue.makeLeftValue("solo")); + plainStore.flush(); + plainStore.close(); + + // Phase 2 — reopen the SAME directory as a HEADERS dual-CF store and read back. + final KeyValueStore>> headersStore = buildHeadersStore(); + final InternalMockProcessorContext ctx2 = newContext(); + headersStore.init(ctx2, headersStore); + try { + final List>>> all = + toListAndCloseIterator(headersStore.all()); + + assertEquals(3, all.size(), "all values from the old plain store should be readable"); + + // k1 retains both list elements, in insertion order, uncorrupted. + assertEquals("k1", all.get(0).key); + assertEquals("right", all.get(0).value.aggregation().rightValue()); + assertNull(all.get(0).value.aggregation().leftValue()); + assertFalse(all.get(0).value.headers().iterator().hasNext(), "migrated headers should be empty"); + + assertEquals("k1", all.get(1).key); + assertEquals("left", all.get(1).value.aggregation().leftValue()); + + assertEquals("k2", all.get(2).key); + assertEquals("solo", all.get(2).value.aggregation().leftValue()); + } finally { + headersStore.close(); + } + } + + @Test + public void shouldAppendAfterUpgradeAndKeepUniformFormat() { + final KeyValueStore> plainStore = buildPlainStore(); + final InternalMockProcessorContext ctx1 = newContext(); + plainStore.init(ctx1, plainStore); + plainStore.put("k1", LeftOrRightValue.makeRightValue("old")); + plainStore.flush(); + plainStore.close(); + + final KeyValueStore>> headersStore = buildHeadersStore(); + final InternalMockProcessorContext ctx2 = newContext(); + headersStore.init(ctx2, headersStore); + try { + // Append a new headers-format element to a key that still lives in the legacy DEFAULT CF. + headersStore.put("k1", AggregationWithHeaders.makeAllowNullable( + LeftOrRightValue.makeLeftValue("new"), new org.apache.kafka.common.header.internals.RecordHeaders())); + + final List>>> all = + toListAndCloseIterator(headersStore.all()); + + assertEquals(2, all.size()); + assertEquals("old", all.get(0).value.aggregation().rightValue()); // migrated legacy element + assertEquals("new", all.get(1).value.aggregation().leftValue()); // freshly appended element + } finally { + headersStore.close(); + } + } +} From f2829d7fa56cdacc5dc2f96a0cd24a470c8e1810 Mon Sep 17 00:00:00 2001 From: Alieh Saeedi Date: Mon, 27 Jul 2026 20:33:55 +0200 Subject: [PATCH 2/6] KAFKA-20413: Keep the list-value store changelog in the PLAIN element format The local headers-aware list store keeps per-element headers inline, as [headersSize][headers][flag][value]. That format must not reach the changelog: the changelog topic is the only durable copy of the state, so its value format is a permanent compatibility contract. Logging the local bytes verbatim would let an old PLAIN reader -- after a version downgrade, or simply after flipping dsl.store.format back to PLAIN -- consume each element's leading empty-headers 0x00 as the LeftOrRightValue flag, silently reading left values as right ones. Do what the other KIP-1271 changelog stores do and keep the headers out of the value. Because one changelog record holds a whole list, N sets of headers have to share one record-header field, so the stripped [headersSize][headersBytes] prefixes are concatenated into a single self-delimiting blob under a reserved header rather than unpacked into individual RecordHeaders. Each chunk carries its own length, so no element count is needed and the encoding does not depend on header ordering. - ListValueStoreUpgradeUtils: add LIST_VALUE_HEADERS_HEADER_KEY, SplitListBlob, splitHeadersListBlob, joinPlainListBlobWithElementHeaders and elementHeaders. Drop the format marker: a record without the control header is a legacy record, which is exactly the all-empty-prefixes case, so the two restore paths collapse into one. - ChangeLoggingListValueBytesStoreWithHeaders: new; overrides put to log the PLAIN blob and attach the prefix header. Copies the record headers so neither the control header nor the vector clock that ProcessorContextImpl#logChange appends can leak onto the record forwarded downstream. - HeadersAwareListValueStore: new marker so StateManagerUtil picks the list-aware converter. - RecordConverters.rawListValueToHeadersListValue: re-inline the per-element headers on restore. - ListValueStoreBuilder: 5-arg ctor selecting the headers-aware changelogger. Nothing constructs any of this yet -- the DSL wiring follows separately -- so merging this changes no behaviour. Tests for ListValueStoreUpgradeUtils itself follow in a separate tests-only PR. Co-Authored-By: Claude Opus 5 (1M context) --- .../processor/internals/StateManagerUtil.java | 9 +- ...LoggingListValueBytesStoreWithHeaders.java | 80 +++++++ .../internals/HeadersAwareListValueStore.java | 33 +++ .../internals/ListValueStoreBuilder.java | 14 +- .../internals/ListValueStoreUpgradeUtils.java | 172 +++++++++++++++ .../state/internals/RecordConverters.java | 34 +++ .../StateManagerUtilConverterTest.java | 31 +++ ...ingListValueBytesStoreWithHeadersTest.java | 208 ++++++++++++++++++ .../ListValueStoreUpgradeUtilsTest.java | 139 ------------ .../state/internals/RecordConvertersTest.java | 61 +++++ .../RocksDBListValueStoreWithHeadersTest.java | 3 +- 11 files changed, 642 insertions(+), 142 deletions(-) create mode 100644 streams/src/main/java/org/apache/kafka/streams/state/internals/ChangeLoggingListValueBytesStoreWithHeaders.java create mode 100644 streams/src/main/java/org/apache/kafka/streams/state/internals/HeadersAwareListValueStore.java create mode 100644 streams/src/test/java/org/apache/kafka/streams/state/internals/ChangeLoggingListValueBytesStoreWithHeadersTest.java delete mode 100644 streams/src/test/java/org/apache/kafka/streams/state/internals/ListValueStoreUpgradeUtilsTest.java diff --git a/streams/src/main/java/org/apache/kafka/streams/processor/internals/StateManagerUtil.java b/streams/src/main/java/org/apache/kafka/streams/processor/internals/StateManagerUtil.java index 42c0da1c51943..ee6b4ccbbde47 100644 --- a/streams/src/main/java/org/apache/kafka/streams/processor/internals/StateManagerUtil.java +++ b/streams/src/main/java/org/apache/kafka/streams/processor/internals/StateManagerUtil.java @@ -27,6 +27,7 @@ import org.apache.kafka.streams.processor.TaskId; import org.apache.kafka.streams.processor.internals.Task.TaskType; import org.apache.kafka.streams.state.SessionStore; +import org.apache.kafka.streams.state.internals.HeadersAwareListValueStore; import org.apache.kafka.streams.state.internals.PlainToHeadersStoreAdapter; import org.apache.kafka.streams.state.internals.PlainToHeadersWindowStoreAdapter; import org.apache.kafka.streams.state.internals.RecordConverter; @@ -41,6 +42,7 @@ import java.util.concurrent.atomic.AtomicReference; import static org.apache.kafka.streams.state.internals.RecordConverters.identity; +import static org.apache.kafka.streams.state.internals.RecordConverters.rawListValueToHeadersListValue; import static org.apache.kafka.streams.state.internals.RecordConverters.rawValueToHeadersValue; import static org.apache.kafka.streams.state.internals.RecordConverters.rawValueToSessionHeadersValue; import static org.apache.kafka.streams.state.internals.RecordConverters.rawValueToTimestampedValue; @@ -75,7 +77,12 @@ static RecordConverter converterForStore(final StateStore store) { // This handles persistent stores that use adapters StateStore current = store; while (current != null) { - if (current instanceof TimestampedToHeadersStoreAdapter || current instanceof TimestampedToHeadersWindowStoreAdapter) { + if (current instanceof HeadersAwareListValueStore) { + // The outer-join ListValueStore is headers-aware but is NOT a HeadersBytesStore: its + // changelog holds whole multi-element list blobs, not single [headers][ts][value] + // payloads, so it needs its own list-aware converter. + return rawListValueToHeadersListValue(); + } else if (current instanceof TimestampedToHeadersStoreAdapter || current instanceof TimestampedToHeadersWindowStoreAdapter) { // Adapter wraps a timestamped store, so restore in timestamped format return rawValueToTimestampedValue(); } else if (current instanceof PlainToHeadersStoreAdapter || current instanceof PlainToHeadersWindowStoreAdapter) { diff --git a/streams/src/main/java/org/apache/kafka/streams/state/internals/ChangeLoggingListValueBytesStoreWithHeaders.java b/streams/src/main/java/org/apache/kafka/streams/state/internals/ChangeLoggingListValueBytesStoreWithHeaders.java new file mode 100644 index 0000000000000..a8f3ade4601a2 --- /dev/null +++ b/streams/src/main/java/org/apache/kafka/streams/state/internals/ChangeLoggingListValueBytesStoreWithHeaders.java @@ -0,0 +1,80 @@ +/* + * 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.header.Headers; +import org.apache.kafka.common.header.internals.RecordHeaders; +import org.apache.kafka.common.utils.Bytes; +import org.apache.kafka.streams.state.KeyValueStore; + +/** + * The HEADERS-mode changelog store for the outer-join {@link ListValueStore}. + *

+ * The local store holds {@code [headersSize][headers][flag][value]} per list element, but that format + * must never reach the changelog: the changelog topic is the only durable copy of the state, so its + * value format is a permanent compatibility contract. If we logged the local bytes verbatim, an old + * PLAIN reader — after a version downgrade, or simply after flipping {@code dsl.store.format} back to + * PLAIN — would read each element's leading empty-headers {@code 0x00} as the {@code LeftOrRightValue} + * flag and silently mistake left values for right ones. + *

+ * So this store does what every other KIP-1271 changelog store does (see + * {@link ChangeLoggingTimestampedKeyValueBytesStoreWithHeaders}, which logs + * {@link Utils#rawPlainValue(byte[])}): it keeps the headers out of the value and puts them in a record + * header instead. The list makes that a little more involved — one changelog record holds the whole + * list, so N sets of headers have to share one header field — which is why the stripped prefixes are + * concatenated into a single self-delimiting blob under + * {@link ListValueStoreUpgradeUtils#LIST_VALUE_HEADERS_HEADER_KEY} rather than unpacked into individual + * {@code RecordHeader}s. + *

+ * Implements {@link HeadersAwareListValueStore} purely so {@code StateManagerUtil.converterForStore} + * selects {@link RecordConverters#rawListValueToHeadersListValue()}, which performs the inverse join on + * restore. + */ +public class ChangeLoggingListValueBytesStoreWithHeaders + extends ChangeLoggingListValueBytesStore + implements HeadersAwareListValueStore { + + ChangeLoggingListValueBytesStoreWithHeaders(final KeyValueStore inner) { + super(inner); + } + + @Override + public void put(final Bytes key, final byte[] value) { + wrapped().put(key, value); + // As in the parent, a tombstone deletes the whole list, so there is nothing to read back and + // no per-element headers to carry. + if (value == null) { + log(key, null, internalContext.recordContext().timestamp(), changelogHeaders(null)); + } else { + final ListValueStoreUpgradeUtils.SplitListBlob split = + ListValueStoreUpgradeUtils.splitHeadersListBlob(wrapped().get(key)); + log(key, split.plainListBlob, internalContext.recordContext().timestamp(), changelogHeaders(split.elementHeaders)); + } + } + + private Headers changelogHeaders(final byte[] elementHeaders) { + // Copy, for two reasons: the live record headers are forwarded downstream, so neither our + // control header nor the vector clock that ProcessorContextImpl#logChange appends to whatever + // instance we hand it may leak into them. + final Headers headers = new RecordHeaders(internalContext.recordContext().headers()); + headers.remove(ListValueStoreUpgradeUtils.LIST_VALUE_HEADERS_HEADER_KEY); + if (elementHeaders != null) { + headers.add(ListValueStoreUpgradeUtils.LIST_VALUE_HEADERS_HEADER_KEY, elementHeaders); + } + return headers; + } +} diff --git a/streams/src/main/java/org/apache/kafka/streams/state/internals/HeadersAwareListValueStore.java b/streams/src/main/java/org/apache/kafka/streams/state/internals/HeadersAwareListValueStore.java new file mode 100644 index 0000000000000..21b1457d2edfe --- /dev/null +++ b/streams/src/main/java/org/apache/kafka/streams/state/internals/HeadersAwareListValueStore.java @@ -0,0 +1,33 @@ +/* + * 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.processor.StateStore; + +/** + * Marker interface for the HEADERS-format outer-join {@link ListValueStore} changelog wrapper. + *

+ * Used solely by {@code StateManagerUtil.converterForStore} to select the list-aware restore + * {@link RecordConverters#rawListValueToHeadersListValue() converter}. + *

+ * Note: this is intentionally NOT {@link org.apache.kafka.streams.state.HeadersBytesStore}. That + * interface would make {@code WrappedStateStore.isHeadersAware} true and wrongly select + * {@code rawValueToHeadersValue()}, which reconstructs a single {@code [headers][ts][value]} payload + * and would corrupt the multi-element list blob used here. + */ +public interface HeadersAwareListValueStore extends StateStore { +} diff --git a/streams/src/main/java/org/apache/kafka/streams/state/internals/ListValueStoreBuilder.java b/streams/src/main/java/org/apache/kafka/streams/state/internals/ListValueStoreBuilder.java index 92b4c4d96e95f..9b88df71d74fe 100644 --- a/streams/src/main/java/org/apache/kafka/streams/state/internals/ListValueStoreBuilder.java +++ b/streams/src/main/java/org/apache/kafka/streams/state/internals/ListValueStoreBuilder.java @@ -26,15 +26,25 @@ public class ListValueStoreBuilder extends AbstractStoreBuilder> { private final KeyValueBytesStoreSupplier storeSupplier; + private final boolean headersFormat; public ListValueStoreBuilder(final KeyValueBytesStoreSupplier storeSupplier, final Serde keySerde, final Serde valueSerde, final Time time) { + this(storeSupplier, keySerde, valueSerde, time, false); + } + + public ListValueStoreBuilder(final KeyValueBytesStoreSupplier storeSupplier, + final Serde keySerde, + final Serde valueSerde, + final Time time, + final boolean headersFormat) { super(storeSupplier.name(), keySerde, valueSerde, time); Objects.requireNonNull(storeSupplier, "storeSupplier can't be null"); Objects.requireNonNull(storeSupplier.metricsScope(), "storeSupplier's metricsScope can't be null"); this.storeSupplier = storeSupplier; + this.headersFormat = headersFormat; } @Override @@ -58,6 +68,8 @@ private KeyValueStore maybeWrapLogging(final KeyValueStore + * The HEADERS element format above is the local, on-disk format only. As everywhere else in + * KIP-1271, the changelog value must keep the pre-headers format so that downgrading — either to an + * older version or just by flipping {@code dsl.store.format} back to PLAIN — can still decode it. The + * {@link #splitHeadersListBlob(byte[]) split} / {@link #joinPlainListBlobWithElementHeaders(byte[], byte[]) join} + * pair moves the per-element headers between the value bytes and a reserved record header for that + * purpose; {@link #LIST_VALUE_HEADERS_HEADER_KEY} documents the wire encoding. */ final class ListValueStoreUpgradeUtils { + /** + * Reserved changelog record-header key carrying the per-element headers of a HEADERS-format list, + * so that the changelog value can stay in the format an old PLAIN store understands. + *

+ * Its value is the concatenation of the {@code [headersSize(varint)][headersBytes]} prefixes that + * were stripped off the list elements, in list order. Each chunk carries its own length, so the + * blob is self-delimiting and no element count is needed. An element with no headers contributes + * a single {@code 0x00} byte. + *

+ * Deliberately namespaced to avoid colliding with user headers that ride along on the record. + * A record without this header is a legacy PLAIN record: see + * {@link #joinPlainListBlobWithElementHeaders(byte[], byte[])}. + */ + static final String LIST_VALUE_HEADERS_HEADER_KEY = "__kafka_streams_list_value_headers__"; + + // The prefix of an element with no headers: headersSize = varint(0), no headers bytes. + private static final byte[] EMPTY_HEADERS_PREFIX = {(byte) 0}; + // Must match ListValueStore.LIST_SERDE. @SuppressWarnings("unchecked") private static final Serde> LIST_SERDE = Serdes.ListSerde(ArrayList.class, Serdes.ByteArray()); @@ -63,4 +94,145 @@ static byte[] convertPlainListBlobToHeadersListBlob(final byte[] plainListBlob) } return LIST_SERDE.serializer().serialize(null, headersElements); } + + /** + * A HEADERS list blob taken apart for the changelog: the value bytes an old PLAIN store can still + * read, plus the per-element headers prefixes to park in {@link #LIST_VALUE_HEADERS_HEADER_KEY}. + */ + static final class SplitListBlob { + final byte[] plainListBlob; + final byte[] elementHeaders; + + SplitListBlob(final byte[] plainListBlob, final byte[] elementHeaders) { + this.plainListBlob = plainListBlob; + this.elementHeaders = elementHeaders; + } + } + + /** + * Splits a HEADERS list blob into the PLAIN list blob plus the concatenated per-element headers + * prefixes. Inverse of {@link #joinPlainListBlobWithElementHeaders(byte[], byte[])}. + *

+ * This is the list-aware counterpart of {@link Utils#rawPlainValue(byte[])}: it keeps the changelog + * value in the pre-headers format so that an old PLAIN store — or a store whose + * {@code dsl.store.format} was flipped back to PLAIN — can still decode it. + * + * @param headersListBlob a {@code ListSerde} blob of {@code [headersSize][headers][flag][value]} + * elements, or {@code null} for a whole-list tombstone + */ + static SplitListBlob splitHeadersListBlob(final byte[] headersListBlob) { + if (headersListBlob == null) { + return new SplitListBlob(null, null); + } + final List headersElements = LIST_SERDE.deserializer().deserialize(null, headersListBlob); + final List plainElements = new ArrayList<>(headersElements.size()); + final ByteArrayOutputStream elementHeaders = new ByteArrayOutputStream(); + + for (final byte[] element : headersElements) { + if (element == null) { + // ListValueStore never appends null, but ListSerde can hold nulls, so keep the pair + // total: a null element round-trips as null and consumes an empty-headers prefix. + plainElements.add(null); + elementHeaders.write(EMPTY_HEADERS_PREFIX, 0, EMPTY_HEADERS_PREFIX.length); + continue; + } + final int prefixLength = headersPrefixLength(element); + elementHeaders.write(element, 0, prefixLength); + final byte[] plainElement = new byte[element.length - prefixLength]; + System.arraycopy(element, prefixLength, plainElement, 0, plainElement.length); + plainElements.add(plainElement); + } + + return new SplitListBlob( + LIST_SERDE.serializer().serialize(null, plainElements), + elementHeaders.toByteArray() + ); + } + + /** + * Rebuilds a HEADERS list blob by re-inlining each element's headers prefix. Inverse of + * {@link #splitHeadersListBlob(byte[])}, and the restore-time counterpart of the split. + * + * @param plainListBlob a {@code ListSerde} blob of {@code [flag][value]} elements, or {@code null} + * @param elementHeaders the concatenated prefixes written by the split, or {@code null}/empty for a + * legacy record that predates the headers format — in which case every element + * gets empty headers, i.e. exactly + * {@link #convertPlainListBlobToHeadersListBlob(byte[])} + */ + static byte[] joinPlainListBlobWithElementHeaders(final byte[] plainListBlob, final byte[] elementHeaders) { + if (plainListBlob == null) { + return null; + } + // Every element contributes at least the one-byte headersSize varint, so an absent or empty + // prefix blob can only mean "legacy record" or "empty list" — both are the all-empty case. + if (elementHeaders == null || elementHeaders.length == 0) { + return convertPlainListBlobToHeadersListBlob(plainListBlob); + } + + final List plainElements = LIST_SERDE.deserializer().deserialize(null, plainListBlob); + final List headersElements = new ArrayList<>(plainElements.size()); + final ByteBuffer prefixes = ByteBuffer.wrap(elementHeaders); + + for (final byte[] plainElement : plainElements) { + final byte[] prefix = readNextHeadersPrefix(prefixes); + if (plainElement == null) { + headersElements.add(null); + continue; + } + final byte[] headersElement = new byte[prefix.length + plainElement.length]; + System.arraycopy(prefix, 0, headersElement, 0, prefix.length); + System.arraycopy(plainElement, 0, headersElement, prefix.length, plainElement.length); + headersElements.add(headersElement); + } + + if (prefixes.hasRemaining()) { + throw new SerializationException("Invalid list-value headers: " + prefixes.remaining() + + " trailing bytes after " + plainElements.size() + " list elements"); + } + return LIST_SERDE.serializer().serialize(null, headersElements); + } + + /** + * @return the per-element headers prefixes carried by a changelog record, or {@code null} if the + * record has none — i.e. it is a legacy record written before the headers format + */ + static byte[] elementHeaders(final Headers headers) { + if (headers == null) { + return null; + } + final Header header = headers.lastHeader(LIST_VALUE_HEADERS_HEADER_KEY); + return header == null ? null : header.value(); + } + + /** + * @return the length of the {@code [headersSize(varint)][headersBytes]} prefix of a HEADERS element + */ + private static int headersPrefixLength(final byte[] headersElement) { + final ByteBuffer buffer = ByteBuffer.wrap(headersElement); + final int headersSize = ByteUtils.readVarint(buffer); + if (headersSize < 0 || headersSize > buffer.remaining()) { + throw new SerializationException("Invalid headers size " + headersSize + " in list element of " + + headersElement.length + " bytes"); + } + return buffer.position() + headersSize; + } + + /** + * Reads one self-delimiting {@code [headersSize(varint)][headersBytes]} chunk, advancing the buffer. + */ + private static byte[] readNextHeadersPrefix(final ByteBuffer prefixes) { + if (!prefixes.hasRemaining()) { + throw new SerializationException( + "Invalid list-value headers: fewer headers prefixes than list elements"); + } + final int start = prefixes.position(); + final int headersSize = ByteUtils.readVarint(prefixes); + if (headersSize < 0 || headersSize > prefixes.remaining()) { + throw new SerializationException("Invalid list-value headers: headers size " + headersSize + + " but only " + prefixes.remaining() + " bytes remaining"); + } + final int end = prefixes.position() + headersSize; + prefixes.position(start); + return Utils.readBytes(prefixes, end - start); + } } diff --git a/streams/src/main/java/org/apache/kafka/streams/state/internals/RecordConverters.java b/streams/src/main/java/org/apache/kafka/streams/state/internals/RecordConverters.java index c040e831f8b8b..338c7c4cf5b84 100644 --- a/streams/src/main/java/org/apache/kafka/streams/state/internals/RecordConverters.java +++ b/streams/src/main/java/org/apache/kafka/streams/state/internals/RecordConverters.java @@ -105,6 +105,40 @@ public static RecordConverter rawValueToSessionHeadersValue() { return RAW_TO_SESSION_WITH_HEADERS_INSTANCE; } + private static final RecordConverter RAW_LIST_TO_HEADERS_LIST_INSTANCE = record -> { + // The outer-join ListValueStore changelog stores the whole list blob, always in the PLAIN + // element format, with the per-element headers parked in a reserved record header. Restoring + // means re-inlining them. Legacy records written before the headers format simply lack that + // header, which is the same as "every element has empty headers" — so there is one path, not + // two. A tombstone (null value) is passed through. + if (record.value() == null) { + return record; + } + + final byte[] convertedValue = ListValueStoreUpgradeUtils.joinPlainListBlobWithElementHeaders( + record.value(), + ListValueStoreUpgradeUtils.elementHeaders(record.headers()) + ); + + return new ConsumerRecord<>( + record.topic(), + record.partition(), + record.offset(), + record.timestamp(), + record.timestampType(), + record.serializedKeySize(), + convertedValue.length, + record.key(), + convertedValue, + record.headers(), + record.leaderEpoch() + ); + }; + + public static RecordConverter rawListValueToHeadersListValue() { + return RAW_LIST_TO_HEADERS_LIST_INSTANCE; + } + // privatize the constructor so the class cannot be instantiated (only used for its static members) private RecordConverters() {} diff --git a/streams/src/test/java/org/apache/kafka/streams/processor/internals/StateManagerUtilConverterTest.java b/streams/src/test/java/org/apache/kafka/streams/processor/internals/StateManagerUtilConverterTest.java index 34bf9979cb60d..187c6f0ffab31 100644 --- a/streams/src/test/java/org/apache/kafka/streams/processor/internals/StateManagerUtilConverterTest.java +++ b/streams/src/test/java/org/apache/kafka/streams/processor/internals/StateManagerUtilConverterTest.java @@ -18,6 +18,8 @@ import org.apache.kafka.streams.processor.StateStore; import org.apache.kafka.streams.state.HeadersBytesStore; +import org.apache.kafka.streams.state.internals.ChangeLoggingListValueBytesStore; +import org.apache.kafka.streams.state.internals.ChangeLoggingListValueBytesStoreWithHeaders; import org.apache.kafka.streams.state.internals.InMemoryKeyValueStore; import org.apache.kafka.streams.state.internals.InMemorySessionStore; import org.apache.kafka.streams.state.internals.InMemoryWindowStore; @@ -43,6 +45,7 @@ import org.mockito.quality.Strictness; import static org.apache.kafka.streams.state.internals.RecordConverters.identity; +import static org.apache.kafka.streams.state.internals.RecordConverters.rawListValueToHeadersListValue; import static org.apache.kafka.streams.state.internals.RecordConverters.rawValueToHeadersValue; import static org.apache.kafka.streams.state.internals.RecordConverters.rawValueToSessionHeadersValue; import static org.apache.kafka.streams.state.internals.RecordConverters.rawValueToTimestampedValue; @@ -191,6 +194,34 @@ public void shouldReturnTimestampedConverterForTimestampedToHeadersPersistentWin assertEquals(rawValueToTimestampedValue(), converter); } + @Test + @SuppressWarnings("unchecked") + public void shouldReturnListConverterForHeadersListValueStore() { + // Metered -> ChangeLoggingListValueBytesStoreWithHeaders (the HeadersAwareListValueStore marker) + final StateStore mockInner = mock(ChangeLoggingListValueBytesStoreWithHeaders.class); + final WrappedStateStore mockMetered = mock(WrappedStateStore.class); + + doReturn(mockInner).when(mockMetered).wrapped(); + + final RecordConverter converter = StateManagerUtil.converterForStore(mockMetered); + + assertEquals(rawListValueToHeadersListValue(), converter); + } + + @Test + @SuppressWarnings("unchecked") + public void shouldReturnIdentityConverterForPlainListValueStore() { + // Metered -> ChangeLoggingListValueBytesStore (no headers marker) + final StateStore mockInner = mock(ChangeLoggingListValueBytesStore.class); + final WrappedStateStore mockMetered = mock(WrappedStateStore.class); + + doReturn(mockInner).when(mockMetered).wrapped(); + + final RecordConverter converter = StateManagerUtil.converterForStore(mockMetered); + + assertEquals(identity(), converter); + } + @Test public void shouldReturnIdentityConverterForPlainToHeadersPersistentSessionStore() { // persistent plain session -> headers session diff --git a/streams/src/test/java/org/apache/kafka/streams/state/internals/ChangeLoggingListValueBytesStoreWithHeadersTest.java b/streams/src/test/java/org/apache/kafka/streams/state/internals/ChangeLoggingListValueBytesStoreWithHeadersTest.java new file mode 100644 index 0000000000000..273dc56a44f68 --- /dev/null +++ b/streams/src/test/java/org/apache/kafka/streams/state/internals/ChangeLoggingListValueBytesStoreWithHeadersTest.java @@ -0,0 +1,208 @@ +/* + * 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.clients.consumer.ConsumerRecord; +import org.apache.kafka.common.header.Headers; +import org.apache.kafka.common.header.internals.RecordHeaders; +import org.apache.kafka.common.metrics.Metrics; +import org.apache.kafka.common.record.TimestampType; +import org.apache.kafka.common.serialization.Serde; +import org.apache.kafka.common.serialization.Serdes; +import org.apache.kafka.common.utils.Bytes; +import org.apache.kafka.common.utils.internals.LogContext; +import org.apache.kafka.streams.processor.internals.MockStreamsMetrics; +import org.apache.kafka.streams.state.AggregationWithHeaders; +import org.apache.kafka.test.InternalMockProcessorContext; +import org.apache.kafka.test.MockRecordCollector; +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.nio.charset.StandardCharsets; +import java.util.ArrayList; +import java.util.List; +import java.util.Optional; + +import static org.junit.jupiter.api.Assertions.assertArrayEquals; +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertNotNull; +import static org.junit.jupiter.api.Assertions.assertNull; + +/** + * The changelog format is the part of this store that can never be changed again, so these tests pin + * it down: the value bytes stay in the PLAIN element format an old reader understands, and the + * per-element headers travel in a reserved record header instead. + */ +public class ChangeLoggingListValueBytesStoreWithHeadersTest { + + private static final String TOPIC = "t"; + + @SuppressWarnings("unchecked") + private static final Serde> LIST_SERDE = Serdes.ListSerde(ArrayList.class, Serdes.ByteArray()); + + private final MockRecordCollector collector = new MockRecordCollector(); + private final InMemoryKeyValueStore inner = new InMemoryKeyValueStore("list"); + private final ListValueStore listStore = new ListValueStore(inner); + private final ChangeLoggingListValueBytesStoreWithHeaders store = + new ChangeLoggingListValueBytesStoreWithHeaders(listStore); + + private final LeftOrRightValueSerde plainSerde = + new LeftOrRightValueSerde<>(Serdes.String(), Serdes.String()); + private final AggregationWithHeadersSerde> headersSerde = + new AggregationWithHeadersSerde<>(plainSerde); + + private final Bytes key = Bytes.wrap("k".getBytes(StandardCharsets.UTF_8)); + + private InternalMockProcessorContext context; + + @BeforeEach + public void before() { + context = new InternalMockProcessorContext<>( + TestUtils.tempDirectory(), + Serdes.String(), + Serdes.String(), + collector, + new ThreadCache(new LogContext("testCache "), 0, new MockStreamsMetrics(new Metrics())) + ); + context.setTime(42L); + store.init(context, store); + } + + @AfterEach + public void after() { + store.close(); + } + + private byte[] headersElement(final LeftOrRightValue value, final Headers headers) { + return headersSerde.serializer().serialize(TOPIC, AggregationWithHeaders.make(value, headers)); + } + + private static Headers headers(final String key, final String value) { + return new RecordHeaders().add(key, value.getBytes(StandardCharsets.UTF_8)); + } + + private byte[] loggedValue(final int index) { + return (byte[]) collector.collected().get(index).value(); + } + + private Headers loggedHeaders(final int index) { + return collector.collected().get(index).headers(); + } + + @Test + public void shouldLogElementsAnOldPlainReaderCanStillDecode() { + // The regression this whole encoding exists for: if the local [headersSize][headers][flag][value] + // format reached the changelog, an old PLAIN reader would consume the leading empty-headers 0x00 + // as the LeftOrRightValue flag -- silently turning left values into right ones. + store.put(key, headersElement(LeftOrRightValue.makeLeftValue("left"), headers("A", "1"))); + store.put(key, headersElement(LeftOrRightValue.makeRightValue("right"), new RecordHeaders())); + + // Read the last logged blob back the way a PLAIN store would. + final List elements = LIST_SERDE.deserializer().deserialize(null, loggedValue(1)); + assertEquals(2, elements.size()); + + final LeftOrRightValue first = plainSerde.deserializer().deserialize(TOPIC, new RecordHeaders(), elements.get(0)); + assertEquals("left", first.leftValue()); + assertNull(first.rightValue(), "a left value must not come back as a right value"); + + final LeftOrRightValue second = plainSerde.deserializer().deserialize(TOPIC, new RecordHeaders(), elements.get(1)); + assertEquals("right", second.rightValue()); + assertNull(second.leftValue()); + } + + @Test + public void shouldRestoreTheLocalBlobFromTheChangelogRecord() { + store.put(key, headersElement(LeftOrRightValue.makeLeftValue("a"), headers("A", "1"))); + store.put(key, headersElement(LeftOrRightValue.makeRightValue("b"), headers("B", "2"))); + + final ConsumerRecord changelogRecord = new ConsumerRecord<>( + "changelog", 0, 0L, 42L, TimestampType.CREATE_TIME, 0, 0, + key.get(), loggedValue(1), loggedHeaders(1), Optional.empty()); + + // What restore reconstructs must be byte-identical to what the local store holds. + assertArrayEquals( + inner.get(key), + RecordConverters.rawListValueToHeadersListValue().convert(changelogRecord).value()); + } + + @Test + public void shouldCarryPerElementHeadersInTheControlHeader() { + store.put(key, headersElement(LeftOrRightValue.makeLeftValue("a"), headers("A", "1"))); + + final byte[] elementHeaders = + ListValueStoreUpgradeUtils.elementHeaders(loggedHeaders(0)); + assertNotNull(elementHeaders, "the per-element headers must be on the record"); + assertArrayEquals( + inner.get(key), + ListValueStoreUpgradeUtils.joinPlainListBlobWithElementHeaders(loggedValue(0), elementHeaders)); + } + + @Test + public void shouldPreserveTheSourceRecordHeadersOnTheChangelogRecord() { + context.headers().add("user-header", "u".getBytes(StandardCharsets.UTF_8)); + + store.put(key, headersElement(LeftOrRightValue.makeLeftValue("a"), headers("A", "1"))); + + assertEquals("u", new String(loggedHeaders(0).lastHeader("user-header").value(), StandardCharsets.UTF_8)); + } + + @Test + public void shouldNotLeakTheControlHeaderIntoTheLiveRecordHeaders() { + // The live headers are forwarded downstream, so nothing we attach to the changelog record may + // show up on them. + store.put(key, headersElement(LeftOrRightValue.makeLeftValue("a"), headers("A", "1"))); + + assertNull(context.headers().lastHeader(ListValueStoreUpgradeUtils.LIST_VALUE_HEADERS_HEADER_KEY)); + } + + @Test + public void shouldLogNullAndNoControlHeaderOnTombstone() { + store.put(key, headersElement(LeftOrRightValue.makeLeftValue("a"), headers("A", "1"))); + store.put(key, null); + + assertEquals(2, collector.collected().size()); + assertNull(loggedValue(1)); + assertNull(ListValueStoreUpgradeUtils.elementHeaders(loggedHeaders(1))); + assertEquals(42L, collector.collected().get(1).timestamp()); + } + + @Test + public void shouldStoreTheHeadersFormatLocally() { + // The split is a changelog concern only: on disk the elements keep their inline headers. + final byte[] element = headersElement(LeftOrRightValue.makeLeftValue("a"), headers("A", "1")); + store.put(key, element); + + assertArrayEquals( + LIST_SERDE.serializer().serialize(null, List.of(element)), + inner.get(key)); + } + + @Test + public void shouldLogTheWholeListOnEveryAppend() { + store.put(key, headersElement(LeftOrRightValue.makeLeftValue("a"), new RecordHeaders())); + store.put(key, headersElement(LeftOrRightValue.makeLeftValue("b"), new RecordHeaders())); + + assertEquals(1, LIST_SERDE.deserializer().deserialize(null, loggedValue(0)).size()); + assertEquals(2, LIST_SERDE.deserializer().deserialize(null, loggedValue(1)).size()); + // One 0x00 prefix per element, so the control header grows with the list. + assertArrayEquals(new byte[]{0}, ListValueStoreUpgradeUtils.elementHeaders(loggedHeaders(0))); + assertArrayEquals(new byte[]{0, 0}, ListValueStoreUpgradeUtils.elementHeaders(loggedHeaders(1))); + } +} diff --git a/streams/src/test/java/org/apache/kafka/streams/state/internals/ListValueStoreUpgradeUtilsTest.java b/streams/src/test/java/org/apache/kafka/streams/state/internals/ListValueStoreUpgradeUtilsTest.java deleted file mode 100644 index 4c796d4025a96..0000000000000 --- a/streams/src/test/java/org/apache/kafka/streams/state/internals/ListValueStoreUpgradeUtilsTest.java +++ /dev/null @@ -1,139 +0,0 @@ -/* - * 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.header.Headers; -import org.apache.kafka.common.header.internals.RecordHeaders; -import org.apache.kafka.common.serialization.Serde; -import org.apache.kafka.common.serialization.Serdes; -import org.apache.kafka.streams.state.AggregationWithHeaders; - -import org.junit.jupiter.api.Test; - -import java.util.ArrayList; -import java.util.Arrays; -import java.util.List; - -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; - -public class ListValueStoreUpgradeUtilsTest { - - private static final String TOPIC = "t"; - private static final Headers EMPTY = new RecordHeaders(); - - @SuppressWarnings("unchecked") - private static final Serde> LIST_SERDE = Serdes.ListSerde(ArrayList.class, Serdes.ByteArray()); - - private final LeftOrRightValueSerde plainSerde = - new LeftOrRightValueSerde<>(Serdes.String(), Serdes.String()); - private final AggregationWithHeadersSerde> headersSerde = - new AggregationWithHeadersSerde<>(plainSerde); - - private byte[] plainElement(final LeftOrRightValue value) { - return plainSerde.serializer().serialize(TOPIC, EMPTY, value); - } - - private byte[] plainListBlob(final List> values) { - final List elements = new ArrayList<>(); - for (final LeftOrRightValue v : values) { - elements.add(plainElement(v)); - } - return LIST_SERDE.serializer().serialize(null, elements); - } - - private List>> readHeadersListBlob(final byte[] blob) { - final List elements = LIST_SERDE.deserializer().deserialize(null, blob); - final List>> out = new ArrayList<>(); - for (final byte[] e : elements) { - out.add(headersSerde.deserializer().deserialize(TOPIC, e)); - } - return out; - } - - @Test - public void shouldConvertRightValueWithoutCorruption() { - final byte[] plain = plainListBlob(List.of(LeftOrRightValue.makeRightValue("right"))); - - final byte[] converted = ListValueStoreUpgradeUtils.convertPlainListBlobToHeadersListBlob(plain); - final List>> result = readHeadersListBlob(converted); - - assertEquals(1, result.size()); - // The pre-fix bug silently corrupted right values (right -> ight); assert it's intact now. - assertEquals("right", result.get(0).aggregation().rightValue()); - assertNull(result.get(0).aggregation().leftValue()); - assertFalse(result.get(0).headers().iterator().hasNext(), "lifted headers should be empty"); - } - - @Test - public void shouldConvertLeftValueThatPreviouslyThrew() { - final byte[] plain = plainListBlob(List.of(LeftOrRightValue.makeLeftValue("left"))); - - final byte[] converted = ListValueStoreUpgradeUtils.convertPlainListBlobToHeadersListBlob(plain); - final List>> result = readHeadersListBlob(converted); - - assertEquals(1, result.size()); - assertEquals("left", result.get(0).aggregation().leftValue()); - assertNull(result.get(0).aggregation().rightValue()); - } - - @Test - public void shouldConvertMultiElementMixedList() { - final byte[] plain = plainListBlob(List.of( - LeftOrRightValue.makeLeftValue("a"), - LeftOrRightValue.makeRightValue("b"), - LeftOrRightValue.makeLeftValue("c") - )); - - final List>> result = - readHeadersListBlob(ListValueStoreUpgradeUtils.convertPlainListBlobToHeadersListBlob(plain)); - - assertEquals(3, result.size()); - assertEquals("a", result.get(0).aggregation().leftValue()); - assertEquals("b", result.get(1).aggregation().rightValue()); - assertEquals("c", result.get(2).aggregation().leftValue()); - } - - @Test - public void shouldPassThroughNull() { - assertNull(ListValueStoreUpgradeUtils.convertPlainListBlobToHeadersListBlob(null)); - } - - @Test - public void shouldConvertEmptyListToEmptyList() { - final byte[] emptyBlob = LIST_SERDE.serializer().serialize(null, new ArrayList<>()); - - final byte[] converted = ListValueStoreUpgradeUtils.convertPlainListBlobToHeadersListBlob(emptyBlob); - - assertEquals(0, LIST_SERDE.deserializer().deserialize(null, converted).size()); - } - - @Test - public void convertedElementShouldBePlainElementPrefixedWithZeroByte() { - final byte[] plainEl = plainElement(LeftOrRightValue.makeRightValue("x")); - final byte[] plain = plainListBlob(List.of(LeftOrRightValue.makeRightValue("x"))); - - final List convertedElements = - LIST_SERDE.deserializer().deserialize(null, ListValueStoreUpgradeUtils.convertPlainListBlobToHeadersListBlob(plain)); - - final byte[] expected = new byte[plainEl.length + 1]; - System.arraycopy(plainEl, 0, expected, 1, plainEl.length); // expected[0] == 0x00 - assertTrue(Arrays.equals(expected, convertedElements.get(0))); - } -} diff --git a/streams/src/test/java/org/apache/kafka/streams/state/internals/RecordConvertersTest.java b/streams/src/test/java/org/apache/kafka/streams/state/internals/RecordConvertersTest.java index d28334c1cc154..6757f38852283 100644 --- a/streams/src/test/java/org/apache/kafka/streams/state/internals/RecordConvertersTest.java +++ b/streams/src/test/java/org/apache/kafka/streams/state/internals/RecordConvertersTest.java @@ -20,12 +20,17 @@ import org.apache.kafka.common.header.Headers; import org.apache.kafka.common.header.internals.RecordHeaders; import org.apache.kafka.common.record.TimestampType; +import org.apache.kafka.common.serialization.Serde; +import org.apache.kafka.common.serialization.Serdes; import org.junit.jupiter.api.Test; import java.nio.ByteBuffer; +import java.util.ArrayList; +import java.util.List; import java.util.Optional; +import static org.apache.kafka.streams.state.internals.RecordConverters.rawListValueToHeadersListValue; import static org.apache.kafka.streams.state.internals.RecordConverters.rawValueToHeadersValue; import static org.apache.kafka.streams.state.internals.RecordConverters.rawValueToSessionHeadersValue; import static org.apache.kafka.streams.state.internals.RecordConverters.rawValueToTimestampedValue; @@ -37,6 +42,10 @@ public class RecordConvertersTest { private final RecordConverter timestampedValueConverter = rawValueToTimestampedValue(); private final RecordConverter headersValueConverter = rawValueToHeadersValue(); private final RecordConverter sessionValueConverter = rawValueToSessionHeadersValue(); + private final RecordConverter listValueConverter = rawListValueToHeadersListValue(); + + @SuppressWarnings("unchecked") + private static final Serde> LIST_SERDE = Serdes.ListSerde(ArrayList.class, Serdes.ByteArray()); @Test @@ -45,6 +54,58 @@ public void shouldPreserveNullValueOnConversion() { assertNull(timestampedValueConverter.convert(nullValueRecord).value()); assertNull(headersValueConverter.convert(nullValueRecord).value()); assertNull(sessionValueConverter.convert(nullValueRecord).value()); + assertNull(listValueConverter.convert(nullValueRecord).value()); + } + + @Test + public void shouldGiveEveryElementEmptyHeadersWhenListValueHeadersAbsent() { + // A legacy record, written before the headers format existed: no control header at all. + final byte[] plainBlob = LIST_SERDE.serializer().serialize(null, List.of(new byte[]{1, 42})); + final ConsumerRecord inputRecord = new ConsumerRecord<>( + "topic", 1, 0, 0, TimestampType.CREATE_TIME, 0, 0, new byte[0], plainBlob, + new RecordHeaders(), Optional.empty()); + + final byte[] converted = listValueConverter.convert(inputRecord).value(); + + // Each element gained a leading 0x00 empty-headers prefix. + assertArrayEquals( + ListValueStoreUpgradeUtils.convertPlainListBlobToHeadersListBlob(plainBlob), + converted); + } + + @Test + public void shouldReInlineListValueHeadersWhenPresent() { + final byte[] plainBlob = LIST_SERDE.serializer().serialize(null, List.of(new byte[]{1, 42}, new byte[]{0, 7})); + // Element 0 carries a 3-byte headers section, element 1 none. headersSize is a zigzag varint, + // so 3 encodes as 6 and 0 encodes as 0. + final byte[] elementHeaders = {6, 9, 9, 9, 0}; + final Headers headers = new RecordHeaders(); + headers.add(ListValueStoreUpgradeUtils.LIST_VALUE_HEADERS_HEADER_KEY, elementHeaders); + final ConsumerRecord inputRecord = new ConsumerRecord<>( + "topic", 1, 0, 0, TimestampType.CREATE_TIME, 0, 0, new byte[0], plainBlob, + headers, Optional.empty()); + + final List converted = + LIST_SERDE.deserializer().deserialize(null, listValueConverter.convert(inputRecord).value()); + + assertArrayEquals(new byte[]{6, 9, 9, 9, 1, 42}, converted.get(0)); + assertArrayEquals(new byte[]{0, 0, 7}, converted.get(1)); + } + + @Test + public void shouldRoundTripListValueHeadersThroughSplitAndJoin() { + final byte[] headersBlob = LIST_SERDE.serializer().serialize(null, + List.of(new byte[]{6, 9, 9, 9, 1, 42}, new byte[]{0, 0, 7})); + + final ListValueStoreUpgradeUtils.SplitListBlob split = + ListValueStoreUpgradeUtils.splitHeadersListBlob(headersBlob); + final Headers headers = new RecordHeaders(); + headers.add(ListValueStoreUpgradeUtils.LIST_VALUE_HEADERS_HEADER_KEY, split.elementHeaders); + final ConsumerRecord inputRecord = new ConsumerRecord<>( + "topic", 1, 0, 0, TimestampType.CREATE_TIME, 0, 0, new byte[0], split.plainListBlob, + headers, Optional.empty()); + + assertArrayEquals(headersBlob, listValueConverter.convert(inputRecord).value()); } @Test diff --git a/streams/src/test/java/org/apache/kafka/streams/state/internals/RocksDBListValueStoreWithHeadersTest.java b/streams/src/test/java/org/apache/kafka/streams/state/internals/RocksDBListValueStoreWithHeadersTest.java index 0bde6cc0c84cb..ed207132bcf79 100644 --- a/streams/src/test/java/org/apache/kafka/streams/state/internals/RocksDBListValueStoreWithHeadersTest.java +++ b/streams/src/test/java/org/apache/kafka/streams/state/internals/RocksDBListValueStoreWithHeadersTest.java @@ -76,7 +76,8 @@ private KeyValueStore(new LeftOrRightValueSerde<>(Serdes.String(), Serdes.String())), - Time.SYSTEM) + Time.SYSTEM, + true) .withCachingDisabled() .withLoggingDisabled() .build(); From fe105d5003db66321dcc34d69d23f09beb14ae05 Mon Sep 17 00:00:00 2001 From: Alieh Saeedi Date: Tue, 28 Jul 2026 19:47:42 +0200 Subject: [PATCH 3/6] address review comments --- .../processor/internals/StateManagerUtil.java | 15 ++++--- ...LoggingListValueBytesStoreWithHeaders.java | 21 +++++---- .../internals/HeadersAwareListValueStore.java | 18 +++----- .../state/internals/ListValueStore.java | 4 +- .../internals/ListValueStoreBuilder.java | 14 +++--- .../internals/ListValueStoreUpgradeUtils.java | 43 ++++++++++--------- .../state/internals/RecordConverters.java | 11 ++++- ...sDBListValueHeadersBytesStoreSupplier.java | 7 ++- .../RocksDBListValueStoreWithHeaders.java | 9 ++-- .../state/internals/WrappedStateStore.java | 14 ++++++ .../StateManagerUtilConverterTest.java | 12 ++++-- ...ingListValueBytesStoreWithHeadersTest.java | 21 ++++----- .../state/internals/RecordConvertersTest.java | 7 +-- .../RocksDBListValueStoreWithHeadersTest.java | 3 +- 14 files changed, 112 insertions(+), 87 deletions(-) diff --git a/streams/src/main/java/org/apache/kafka/streams/processor/internals/StateManagerUtil.java b/streams/src/main/java/org/apache/kafka/streams/processor/internals/StateManagerUtil.java index ee6b4ccbbde47..3134946268339 100644 --- a/streams/src/main/java/org/apache/kafka/streams/processor/internals/StateManagerUtil.java +++ b/streams/src/main/java/org/apache/kafka/streams/processor/internals/StateManagerUtil.java @@ -27,7 +27,6 @@ import org.apache.kafka.streams.processor.TaskId; import org.apache.kafka.streams.processor.internals.Task.TaskType; import org.apache.kafka.streams.state.SessionStore; -import org.apache.kafka.streams.state.internals.HeadersAwareListValueStore; import org.apache.kafka.streams.state.internals.PlainToHeadersStoreAdapter; import org.apache.kafka.streams.state.internals.PlainToHeadersWindowStoreAdapter; import org.apache.kafka.streams.state.internals.RecordConverter; @@ -47,6 +46,7 @@ import static org.apache.kafka.streams.state.internals.RecordConverters.rawValueToSessionHeadersValue; import static org.apache.kafka.streams.state.internals.RecordConverters.rawValueToTimestampedValue; import static org.apache.kafka.streams.state.internals.WrappedStateStore.isHeadersAware; +import static org.apache.kafka.streams.state.internals.WrappedStateStore.isHeadersAwareListValue; import static org.apache.kafka.streams.state.internals.WrappedStateStore.isTimestamped; import static org.apache.kafka.streams.state.internals.WrappedStateStore.isVersioned; @@ -61,6 +61,12 @@ final class StateManagerUtil { private StateManagerUtil() {} static RecordConverter converterForStore(final StateStore store) { + // The outer-join ListValueStore is headers-aware, but one changelog record holds a whole + // multi-element list blob rather than a single [headers][ts][value] payload, so it needs its own + // converter. It is also a HeadersBytesStore, hence this check has to come first. + if (isHeadersAwareListValue(store)) { + return rawListValueToHeadersListValue(); + } // First check if the top-level store implements HeadersBytesStore or TimestampedBytesStore if (isHeadersAware(store)) { if (store instanceof SessionStore) { @@ -77,12 +83,7 @@ static RecordConverter converterForStore(final StateStore store) { // This handles persistent stores that use adapters StateStore current = store; while (current != null) { - if (current instanceof HeadersAwareListValueStore) { - // The outer-join ListValueStore is headers-aware but is NOT a HeadersBytesStore: its - // changelog holds whole multi-element list blobs, not single [headers][ts][value] - // payloads, so it needs its own list-aware converter. - return rawListValueToHeadersListValue(); - } else if (current instanceof TimestampedToHeadersStoreAdapter || current instanceof TimestampedToHeadersWindowStoreAdapter) { + if (current instanceof TimestampedToHeadersStoreAdapter || current instanceof TimestampedToHeadersWindowStoreAdapter) { // Adapter wraps a timestamped store, so restore in timestamped format return rawValueToTimestampedValue(); } else if (current instanceof PlainToHeadersStoreAdapter || current instanceof PlainToHeadersWindowStoreAdapter) { diff --git a/streams/src/main/java/org/apache/kafka/streams/state/internals/ChangeLoggingListValueBytesStoreWithHeaders.java b/streams/src/main/java/org/apache/kafka/streams/state/internals/ChangeLoggingListValueBytesStoreWithHeaders.java index a8f3ade4601a2..ce5c02a148da0 100644 --- a/streams/src/main/java/org/apache/kafka/streams/state/internals/ChangeLoggingListValueBytesStoreWithHeaders.java +++ b/streams/src/main/java/org/apache/kafka/streams/state/internals/ChangeLoggingListValueBytesStoreWithHeaders.java @@ -40,13 +40,11 @@ * {@link ListValueStoreUpgradeUtils#LIST_VALUE_HEADERS_HEADER_KEY} rather than unpacked into individual * {@code RecordHeader}s. *

- * Implements {@link HeadersAwareListValueStore} purely so {@code StateManagerUtil.converterForStore} - * selects {@link RecordConverters#rawListValueToHeadersListValue()}, which performs the inverse join on - * restore. + * The inverse join on restore is performed by {@link RecordConverters#rawListValueToHeadersListValue()}, + * which {@code StateManagerUtil.converterForStore} selects off the {@link HeadersAwareListValueStore} + * marker on the bytes store below. */ -public class ChangeLoggingListValueBytesStoreWithHeaders - extends ChangeLoggingListValueBytesStore - implements HeadersAwareListValueStore { +public class ChangeLoggingListValueBytesStoreWithHeaders extends ChangeLoggingListValueBytesStore { ChangeLoggingListValueBytesStoreWithHeaders(final KeyValueStore inner) { super(inner); @@ -67,11 +65,12 @@ public void put(final Bytes key, final byte[] value) { } private Headers changelogHeaders(final byte[] elementHeaders) { - // Copy, for two reasons: the live record headers are forwarded downstream, so neither our - // control header nor the vector clock that ProcessorContextImpl#logChange appends to whatever - // instance we hand it may leak into them. - final Headers headers = new RecordHeaders(internalContext.recordContext().headers()); - headers.remove(ListValueStoreUpgradeUtils.LIST_VALUE_HEADERS_HEADER_KEY); + // A fresh instance, never the live record headers: ProcessorContextImpl#logChange appends the + // vector clock to whatever it is handed, which would leak into the record forwarded downstream. + // Nothing is copied from the record context -- the per-element prefixes in the control blob + // already carry this record's own headers -- which also matches the PLAIN parent, and the + // value-with-headers stores, which log the headers taken out of the value. + final Headers headers = new RecordHeaders(); if (elementHeaders != null) { headers.add(ListValueStoreUpgradeUtils.LIST_VALUE_HEADERS_HEADER_KEY, elementHeaders); } diff --git a/streams/src/main/java/org/apache/kafka/streams/state/internals/HeadersAwareListValueStore.java b/streams/src/main/java/org/apache/kafka/streams/state/internals/HeadersAwareListValueStore.java index 21b1457d2edfe..ebcc41dfff3f1 100644 --- a/streams/src/main/java/org/apache/kafka/streams/state/internals/HeadersAwareListValueStore.java +++ b/streams/src/main/java/org/apache/kafka/streams/state/internals/HeadersAwareListValueStore.java @@ -16,18 +16,14 @@ */ package org.apache.kafka.streams.state.internals; -import org.apache.kafka.streams.processor.StateStore; - /** - * Marker interface for the HEADERS-format outer-join {@link ListValueStore} changelog wrapper. - *

- * Used solely by {@code StateManagerUtil.converterForStore} to select the list-aware restore - * {@link RecordConverters#rawListValueToHeadersListValue() converter}. + * Marker interface for the bytes store behind the HEADERS-format outer-join {@link ListValueStore}. *

- * Note: this is intentionally NOT {@link org.apache.kafka.streams.state.HeadersBytesStore}. That - * interface would make {@code WrappedStateStore.isHeadersAware} true and wrongly select - * {@code rawValueToHeadersValue()}, which reconstructs a single {@code [headers][ts][value]} payload - * and would corrupt the multi-element list blob used here. + * Used by {@code StateManagerUtil.converterForStore} to select the list-aware restore + * {@link RecordConverters#rawListValueToHeadersListValue() converter}. Such a store is also a + * {@link org.apache.kafka.streams.state.HeadersBytesStore}, so this marker has to be checked + * first: the generic headers converter reconstructs a single {@code [headers][ts][value]} + * payload, which would corrupt a multi-element list blob. */ -public interface HeadersAwareListValueStore extends StateStore { +public interface HeadersAwareListValueStore { } diff --git a/streams/src/main/java/org/apache/kafka/streams/state/internals/ListValueStore.java b/streams/src/main/java/org/apache/kafka/streams/state/internals/ListValueStore.java index 358c2f0e1dfa1..ddb59adf38530 100644 --- a/streams/src/main/java/org/apache/kafka/streams/state/internals/ListValueStore.java +++ b/streams/src/main/java/org/apache/kafka/streams/state/internals/ListValueStore.java @@ -42,7 +42,9 @@ public class ListValueStore extends WrappedStateStore, Bytes, byte[]> implements KeyValueStore { - private static final Serde> LIST_SERDE = Serdes.ListSerde(ArrayList.class, Serdes.ByteArray()); + // Defines the on-disk blob encoding of a list of values. Package-private because the changelog + // split/join in ListValueStoreUpgradeUtils has to read and write the very same encoding. + static final Serde> LIST_SERDE = Serdes.ListSerde(ArrayList.class, Serdes.ByteArray()); ListValueStore(final KeyValueStore bytesStore) { super(bytesStore); diff --git a/streams/src/main/java/org/apache/kafka/streams/state/internals/ListValueStoreBuilder.java b/streams/src/main/java/org/apache/kafka/streams/state/internals/ListValueStoreBuilder.java index 9b88df71d74fe..c64e374dee63a 100644 --- a/streams/src/main/java/org/apache/kafka/streams/state/internals/ListValueStoreBuilder.java +++ b/streams/src/main/java/org/apache/kafka/streams/state/internals/ListValueStoreBuilder.java @@ -19,6 +19,7 @@ import org.apache.kafka.common.serialization.Serde; import org.apache.kafka.common.utils.Bytes; import org.apache.kafka.common.utils.Time; +import org.apache.kafka.streams.state.HeadersBytesStoreSupplier; import org.apache.kafka.streams.state.KeyValueBytesStoreSupplier; import org.apache.kafka.streams.state.KeyValueStore; @@ -32,19 +33,14 @@ public ListValueStoreBuilder(final KeyValueBytesStoreSupplier storeSupplier, final Serde keySerde, final Serde valueSerde, final Time time) { - this(storeSupplier, keySerde, valueSerde, time, false); - } - - public ListValueStoreBuilder(final KeyValueBytesStoreSupplier storeSupplier, - final Serde keySerde, - final Serde valueSerde, - final Time time, - final boolean headersFormat) { super(storeSupplier.name(), keySerde, valueSerde, time); Objects.requireNonNull(storeSupplier, "storeSupplier can't be null"); Objects.requireNonNull(storeSupplier.metricsScope(), "storeSupplier's metricsScope can't be null"); this.storeSupplier = storeSupplier; - this.headersFormat = headersFormat; + // The element format follows the bytes store: a headers-aware supplier means each list element + // carries its own headers, which in turn is what the changelogger has to strip out before + // logging. Inferring it here keeps the two from ever disagreeing. + this.headersFormat = storeSupplier instanceof HeadersBytesStoreSupplier; } @Override diff --git a/streams/src/main/java/org/apache/kafka/streams/state/internals/ListValueStoreUpgradeUtils.java b/streams/src/main/java/org/apache/kafka/streams/state/internals/ListValueStoreUpgradeUtils.java index daf6e8910560b..de20b8c7d2630 100644 --- a/streams/src/main/java/org/apache/kafka/streams/state/internals/ListValueStoreUpgradeUtils.java +++ b/streams/src/main/java/org/apache/kafka/streams/state/internals/ListValueStoreUpgradeUtils.java @@ -19,7 +19,6 @@ import org.apache.kafka.common.errors.SerializationException; import org.apache.kafka.common.header.Header; import org.apache.kafka.common.header.Headers; -import org.apache.kafka.common.serialization.Serde; import org.apache.kafka.common.serialization.Serdes; import org.apache.kafka.common.utils.internals.ByteUtils; import org.apache.kafka.streams.state.HeadersBytesStore; @@ -29,6 +28,8 @@ import java.util.ArrayList; import java.util.List; +import static org.apache.kafka.streams.state.internals.ListValueStore.LIST_SERDE; + /** * Helpers for migrating the outer-join {@link ListValueStore} from the pre-headers PLAIN element * format to the HEADERS element format (KIP-1271, added for AK 4.4). @@ -63,18 +64,14 @@ final class ListValueStoreUpgradeUtils { * blob is self-delimiting and no element count is needed. An element with no headers contributes * a single {@code 0x00} byte. *

- * Deliberately namespaced to avoid colliding with user headers that ride along on the record. + * Kept short like the {@code "v"} and {@code "c"} keys Kafka already writes onto changelog records: + * this one is paid on every record, and the changelog record carries no user headers to collide with + * (see {@code ChangeLoggingListValueBytesStoreWithHeaders#changelogHeaders}). + *

* A record without this header is a legacy PLAIN record: see * {@link #joinPlainListBlobWithElementHeaders(byte[], byte[])}. */ - static final String LIST_VALUE_HEADERS_HEADER_KEY = "__kafka_streams_list_value_headers__"; - - // The prefix of an element with no headers: headersSize = varint(0), no headers bytes. - private static final byte[] EMPTY_HEADERS_PREFIX = {(byte) 0}; - - // Must match ListValueStore.LIST_SERDE. - @SuppressWarnings("unchecked") - private static final Serde> LIST_SERDE = Serdes.ListSerde(ArrayList.class, Serdes.ByteArray()); + static final String LIST_VALUE_HEADERS_HEADER_KEY = "vh"; private ListValueStoreUpgradeUtils() {} @@ -127,16 +124,18 @@ static SplitListBlob splitHeadersListBlob(final byte[] headersListBlob) { final List headersElements = LIST_SERDE.deserializer().deserialize(null, headersListBlob); final List plainElements = new ArrayList<>(headersElements.size()); final ByteArrayOutputStream elementHeaders = new ByteArrayOutputStream(); + boolean anyHeaders = false; for (final byte[] element : headersElements) { if (element == null) { - // ListValueStore never appends null, but ListSerde can hold nulls, so keep the pair - // total: a null element round-trips as null and consumes an empty-headers prefix. - plainElements.add(null); - elementHeaders.write(EMPTY_HEADERS_PREFIX, 0, EMPTY_HEADERS_PREFIX.length); - continue; + // ListValueStore is the only writer and never appends null: put/putIfAbsent turn a null + // value into a whole-list delete, and putAll is unsupported. + throw new SerializationException("Unexpected null element in list-value blob of " + + headersElements.size() + " elements"); } final int prefixLength = headersPrefixLength(element); + // An empty headers section is exactly the one-byte varint 0, so anything longer carries headers. + anyHeaders |= prefixLength > 1; elementHeaders.write(element, 0, prefixLength); final byte[] plainElement = new byte[element.length - prefixLength]; System.arraycopy(element, prefixLength, plainElement, 0, plainElement.length); @@ -145,7 +144,10 @@ static SplitListBlob splitHeadersListBlob(final byte[] headersListBlob) { return new SplitListBlob( LIST_SERDE.serializer().serialize(null, plainElements), - elementHeaders.toByteArray() + // All-empty prefixes carry no information: the join side reads an absent blob as "every + // element has empty headers", so dropping it makes the changelog record byte-identical to + // what a PLAIN store would have logged. + anyHeaders ? elementHeaders.toByteArray() : null ); } @@ -176,8 +178,10 @@ static byte[] joinPlainListBlobWithElementHeaders(final byte[] plainListBlob, fi for (final byte[] plainElement : plainElements) { final byte[] prefix = readNextHeadersPrefix(prefixes); if (plainElement == null) { - headersElements.add(null); - continue; + // As in splitHeadersListBlob: ListValueStore never appends null, so a null element here + // means a corrupt or foreign changelog record. + throw new SerializationException("Unexpected null element in list-value changelog record of " + + plainElements.size() + " elements"); } final byte[] headersElement = new byte[prefix.length + plainElement.length]; System.arraycopy(prefix, 0, headersElement, 0, prefix.length); @@ -197,9 +201,6 @@ static byte[] joinPlainListBlobWithElementHeaders(final byte[] plainListBlob, fi * record has none — i.e. it is a legacy record written before the headers format */ static byte[] elementHeaders(final Headers headers) { - if (headers == null) { - return null; - } final Header header = headers.lastHeader(LIST_VALUE_HEADERS_HEADER_KEY); return header == null ? null : header.value(); } diff --git a/streams/src/main/java/org/apache/kafka/streams/state/internals/RecordConverters.java b/streams/src/main/java/org/apache/kafka/streams/state/internals/RecordConverters.java index 338c7c4cf5b84..5576c36dd288a 100644 --- a/streams/src/main/java/org/apache/kafka/streams/state/internals/RecordConverters.java +++ b/streams/src/main/java/org/apache/kafka/streams/state/internals/RecordConverters.java @@ -18,6 +18,7 @@ import org.apache.kafka.clients.consumer.ConsumerRecord; import org.apache.kafka.common.header.Headers; +import org.apache.kafka.common.header.internals.RecordHeaders; import org.apache.kafka.common.utils.internals.ByteUtils; import java.nio.ByteBuffer; @@ -120,6 +121,14 @@ public static RecordConverter rawValueToSessionHeadersValue() { ListValueStoreUpgradeUtils.elementHeaders(record.headers()) ); + // Our control header has done its job once the prefixes are back inside the value, so keep it + // off the restored record. Copy rather than remove in place: the Headers instance belongs to the + // caller, and a converter must not have side effects on its input. Everything else is passed + // through, because the restore path reads the position/vector clock back out of these headers + // (see ChangelogRecordDeserializationHelper#applyChecksAndUpdatePosition). + final Headers headers = new RecordHeaders(record.headers()); + headers.remove(ListValueStoreUpgradeUtils.LIST_VALUE_HEADERS_HEADER_KEY); + return new ConsumerRecord<>( record.topic(), record.partition(), @@ -130,7 +139,7 @@ public static RecordConverter rawValueToSessionHeadersValue() { convertedValue.length, record.key(), convertedValue, - record.headers(), + headers, record.leaderEpoch() ); }; diff --git a/streams/src/main/java/org/apache/kafka/streams/state/internals/RocksDBListValueHeadersBytesStoreSupplier.java b/streams/src/main/java/org/apache/kafka/streams/state/internals/RocksDBListValueHeadersBytesStoreSupplier.java index ba6edad9f8537..af392d54043ee 100644 --- a/streams/src/main/java/org/apache/kafka/streams/state/internals/RocksDBListValueHeadersBytesStoreSupplier.java +++ b/streams/src/main/java/org/apache/kafka/streams/state/internals/RocksDBListValueHeadersBytesStoreSupplier.java @@ -17,14 +17,19 @@ package org.apache.kafka.streams.state.internals; import org.apache.kafka.common.utils.Bytes; +import org.apache.kafka.streams.state.HeadersBytesStoreSupplier; import org.apache.kafka.streams.state.KeyValueBytesStoreSupplier; import org.apache.kafka.streams.state.KeyValueStore; /** * Supplies the persistent, dual-column-family {@link RocksDBListValueStoreWithHeaders} used as the * bytes store for the outer-join {@link ListValueStore} in HEADERS mode. + *

+ * Being a {@link HeadersBytesStoreSupplier} is also what tells {@link ListValueStoreBuilder} to wrap + * the store in the headers-aware changelogger. */ -public class RocksDBListValueHeadersBytesStoreSupplier implements KeyValueBytesStoreSupplier { +public class RocksDBListValueHeadersBytesStoreSupplier + implements KeyValueBytesStoreSupplier, HeadersBytesStoreSupplier { private final String name; diff --git a/streams/src/main/java/org/apache/kafka/streams/state/internals/RocksDBListValueStoreWithHeaders.java b/streams/src/main/java/org/apache/kafka/streams/state/internals/RocksDBListValueStoreWithHeaders.java index 0ccf9dfc4c4d6..a8ee2e0d2a111 100644 --- a/streams/src/main/java/org/apache/kafka/streams/state/internals/RocksDBListValueStoreWithHeaders.java +++ b/streams/src/main/java/org/apache/kafka/streams/state/internals/RocksDBListValueStoreWithHeaders.java @@ -16,6 +16,7 @@ */ package org.apache.kafka.streams.state.internals; +import org.apache.kafka.streams.state.HeadersBytesStore; import org.apache.kafka.streams.state.internals.metrics.RocksDBMetricsRecorder; import org.rocksdb.ColumnFamilyDescriptor; @@ -45,10 +46,12 @@ * {@link ListValueStoreUpgradeUtils#convertPlainListBlobToHeadersListBlob} and migrates it forward. * Otherwise a {@link RocksDBStore.SingleColumnFamilyAccessor} over the headers column family is used. *

- * This class intentionally does NOT implement {@code HeadersBytesStore}; see - * {@link HeadersAwareListValueStore} for why. + * Like its siblings it is a {@link HeadersBytesStore}, and additionally a + * {@link HeadersAwareListValueStore} so that restore picks the list-aware converter rather than the + * generic whole-value one; see {@link HeadersAwareListValueStore}. */ -public class RocksDBListValueStoreWithHeaders extends RocksDBStore { +public class RocksDBListValueStoreWithHeaders extends RocksDBStore + implements HeadersBytesStore, HeadersAwareListValueStore { private static final Logger log = LoggerFactory.getLogger(RocksDBListValueStoreWithHeaders.class); diff --git a/streams/src/main/java/org/apache/kafka/streams/state/internals/WrappedStateStore.java b/streams/src/main/java/org/apache/kafka/streams/state/internals/WrappedStateStore.java index 6032feb362ac7..426ddc854ecd5 100644 --- a/streams/src/main/java/org/apache/kafka/streams/state/internals/WrappedStateStore.java +++ b/streams/src/main/java/org/apache/kafka/streams/state/internals/WrappedStateStore.java @@ -66,6 +66,20 @@ public static boolean isHeadersAware(final StateStore stateStore) { } } + /** + * A headers-aware store whose values are whole list blobs rather than single values. Such a store is + * also {@link #isHeadersAware(StateStore) headers-aware}, so this has to be tested first. + */ + public static boolean isHeadersAwareListValue(final StateStore stateStore) { + if (stateStore instanceof HeadersAwareListValueStore) { + return true; + } else if (stateStore instanceof WrappedStateStore) { + return isHeadersAwareListValue(((WrappedStateStore) stateStore).wrapped()); + } else { + return false; + } + } + private final S wrapped; public WrappedStateStore(final S wrapped) { diff --git a/streams/src/test/java/org/apache/kafka/streams/processor/internals/StateManagerUtilConverterTest.java b/streams/src/test/java/org/apache/kafka/streams/processor/internals/StateManagerUtilConverterTest.java index 187c6f0ffab31..cbbb9f2a1a672 100644 --- a/streams/src/test/java/org/apache/kafka/streams/processor/internals/StateManagerUtilConverterTest.java +++ b/streams/src/test/java/org/apache/kafka/streams/processor/internals/StateManagerUtilConverterTest.java @@ -19,7 +19,6 @@ import org.apache.kafka.streams.processor.StateStore; import org.apache.kafka.streams.state.HeadersBytesStore; import org.apache.kafka.streams.state.internals.ChangeLoggingListValueBytesStore; -import org.apache.kafka.streams.state.internals.ChangeLoggingListValueBytesStoreWithHeaders; import org.apache.kafka.streams.state.internals.InMemoryKeyValueStore; import org.apache.kafka.streams.state.internals.InMemorySessionStore; import org.apache.kafka.streams.state.internals.InMemoryWindowStore; @@ -32,6 +31,7 @@ import org.apache.kafka.streams.state.internals.PlainToHeadersStoreAdapter; import org.apache.kafka.streams.state.internals.PlainToHeadersWindowStoreAdapter; import org.apache.kafka.streams.state.internals.RecordConverter; +import org.apache.kafka.streams.state.internals.RocksDBListValueStoreWithHeaders; import org.apache.kafka.streams.state.internals.SessionToHeadersStoreAdapter; import org.apache.kafka.streams.state.internals.TimestampedToHeadersStoreAdapter; import org.apache.kafka.streams.state.internals.TimestampedToHeadersWindowStoreAdapter; @@ -197,11 +197,15 @@ public void shouldReturnTimestampedConverterForTimestampedToHeadersPersistentWin @Test @SuppressWarnings("unchecked") public void shouldReturnListConverterForHeadersListValueStore() { - // Metered -> ChangeLoggingListValueBytesStoreWithHeaders (the HeadersAwareListValueStore marker) - final StateStore mockInner = mock(ChangeLoggingListValueBytesStoreWithHeaders.class); + // Metered -> ChangeLogging -> RocksDBListValueStoreWithHeaders, which carries the + // HeadersAwareListValueStore marker. It is also a HeadersBytesStore, so this pins the branch + // order in converterForStore: the list converter must win over the whole-value one. + final StateStore mockInner = mock(RocksDBListValueStoreWithHeaders.class); + final WrappedStateStore mockChangeLogging = mock(WrappedStateStore.class); final WrappedStateStore mockMetered = mock(WrappedStateStore.class); - doReturn(mockInner).when(mockMetered).wrapped(); + doReturn(mockInner).when(mockChangeLogging).wrapped(); + doReturn(mockChangeLogging).when(mockMetered).wrapped(); final RecordConverter converter = StateManagerUtil.converterForStore(mockMetered); diff --git a/streams/src/test/java/org/apache/kafka/streams/state/internals/ChangeLoggingListValueBytesStoreWithHeadersTest.java b/streams/src/test/java/org/apache/kafka/streams/state/internals/ChangeLoggingListValueBytesStoreWithHeadersTest.java index 273dc56a44f68..1efc1f87b8127 100644 --- a/streams/src/test/java/org/apache/kafka/streams/state/internals/ChangeLoggingListValueBytesStoreWithHeadersTest.java +++ b/streams/src/test/java/org/apache/kafka/streams/state/internals/ChangeLoggingListValueBytesStoreWithHeadersTest.java @@ -21,7 +21,6 @@ import org.apache.kafka.common.header.internals.RecordHeaders; import org.apache.kafka.common.metrics.Metrics; import org.apache.kafka.common.record.TimestampType; -import org.apache.kafka.common.serialization.Serde; import org.apache.kafka.common.serialization.Serdes; import org.apache.kafka.common.utils.Bytes; import org.apache.kafka.common.utils.internals.LogContext; @@ -36,10 +35,10 @@ import org.junit.jupiter.api.Test; import java.nio.charset.StandardCharsets; -import java.util.ArrayList; import java.util.List; import java.util.Optional; +import static org.apache.kafka.streams.state.internals.ListValueStore.LIST_SERDE; import static org.junit.jupiter.api.Assertions.assertArrayEquals; import static org.junit.jupiter.api.Assertions.assertEquals; import static org.junit.jupiter.api.Assertions.assertNotNull; @@ -54,9 +53,6 @@ public class ChangeLoggingListValueBytesStoreWithHeadersTest { private static final String TOPIC = "t"; - @SuppressWarnings("unchecked") - private static final Serde> LIST_SERDE = Serdes.ListSerde(ArrayList.class, Serdes.ByteArray()); - private final MockRecordCollector collector = new MockRecordCollector(); private final InMemoryKeyValueStore inner = new InMemoryKeyValueStore("list"); private final ListValueStore listStore = new ListValueStore(inner); @@ -155,12 +151,16 @@ public void shouldCarryPerElementHeadersInTheControlHeader() { } @Test - public void shouldPreserveTheSourceRecordHeadersOnTheChangelogRecord() { + public void shouldLogOnlyTheControlHeaderAndNotCopyTheSourceRecordHeaders() { + // The changelog record's headers carry the store's own data, nothing from the record context: + // this record's user headers are already inside the control blob, as the prefix of the element + // just appended. The PLAIN parent logs empty headers for the same reason. context.headers().add("user-header", "u".getBytes(StandardCharsets.UTF_8)); store.put(key, headersElement(LeftOrRightValue.makeLeftValue("a"), headers("A", "1"))); - assertEquals("u", new String(loggedHeaders(0).lastHeader("user-header").value(), StandardCharsets.UTF_8)); + assertNull(loggedHeaders(0).lastHeader("user-header")); + assertNotNull(loggedHeaders(0).lastHeader(ListValueStoreUpgradeUtils.LIST_VALUE_HEADERS_HEADER_KEY)); } @Test @@ -201,8 +201,9 @@ public void shouldLogTheWholeListOnEveryAppend() { assertEquals(1, LIST_SERDE.deserializer().deserialize(null, loggedValue(0)).size()); assertEquals(2, LIST_SERDE.deserializer().deserialize(null, loggedValue(1)).size()); - // One 0x00 prefix per element, so the control header grows with the list. - assertArrayEquals(new byte[]{0}, ListValueStoreUpgradeUtils.elementHeaders(loggedHeaders(0))); - assertArrayEquals(new byte[]{0, 0}, ListValueStoreUpgradeUtils.elementHeaders(loggedHeaders(1))); + // No element has headers, so no control header is written at all -- these records are exactly + // what a PLAIN store would have logged. + assertNull(ListValueStoreUpgradeUtils.elementHeaders(loggedHeaders(0))); + assertNull(ListValueStoreUpgradeUtils.elementHeaders(loggedHeaders(1))); } } diff --git a/streams/src/test/java/org/apache/kafka/streams/state/internals/RecordConvertersTest.java b/streams/src/test/java/org/apache/kafka/streams/state/internals/RecordConvertersTest.java index 6757f38852283..d2f809de5be31 100644 --- a/streams/src/test/java/org/apache/kafka/streams/state/internals/RecordConvertersTest.java +++ b/streams/src/test/java/org/apache/kafka/streams/state/internals/RecordConvertersTest.java @@ -20,16 +20,14 @@ import org.apache.kafka.common.header.Headers; import org.apache.kafka.common.header.internals.RecordHeaders; import org.apache.kafka.common.record.TimestampType; -import org.apache.kafka.common.serialization.Serde; -import org.apache.kafka.common.serialization.Serdes; import org.junit.jupiter.api.Test; import java.nio.ByteBuffer; -import java.util.ArrayList; import java.util.List; import java.util.Optional; +import static org.apache.kafka.streams.state.internals.ListValueStore.LIST_SERDE; import static org.apache.kafka.streams.state.internals.RecordConverters.rawListValueToHeadersListValue; import static org.apache.kafka.streams.state.internals.RecordConverters.rawValueToHeadersValue; import static org.apache.kafka.streams.state.internals.RecordConverters.rawValueToSessionHeadersValue; @@ -44,9 +42,6 @@ public class RecordConvertersTest { private final RecordConverter sessionValueConverter = rawValueToSessionHeadersValue(); private final RecordConverter listValueConverter = rawListValueToHeadersListValue(); - @SuppressWarnings("unchecked") - private static final Serde> LIST_SERDE = Serdes.ListSerde(ArrayList.class, Serdes.ByteArray()); - @Test public void shouldPreserveNullValueOnConversion() { diff --git a/streams/src/test/java/org/apache/kafka/streams/state/internals/RocksDBListValueStoreWithHeadersTest.java b/streams/src/test/java/org/apache/kafka/streams/state/internals/RocksDBListValueStoreWithHeadersTest.java index ed207132bcf79..0bde6cc0c84cb 100644 --- a/streams/src/test/java/org/apache/kafka/streams/state/internals/RocksDBListValueStoreWithHeadersTest.java +++ b/streams/src/test/java/org/apache/kafka/streams/state/internals/RocksDBListValueStoreWithHeadersTest.java @@ -76,8 +76,7 @@ private KeyValueStore(new LeftOrRightValueSerde<>(Serdes.String(), Serdes.String())), - Time.SYSTEM, - true) + Time.SYSTEM) .withCachingDisabled() .withLoggingDisabled() .build(); From 7844c656b331419be6b0485699839520dc08ac6c Mon Sep 17 00:00:00 2001 From: Alieh Saeedi Date: Mon, 27 Jul 2026 20:58:55 +0200 Subject: [PATCH 4/6] KAFKA-20413: Make the stream-stream outer join use the headers-aware list store Wires up the storage and changelog layers added in the previous PR, and carries per-record headers through the non-joined (spurious-result) emit. - OuterStreamJoinStoreFactory: in HEADERS mode, wrap the value serde in AggregationWithHeadersSerde and build the store through the headers-aware ListValueStoreBuilder, picking the dual-column-family supplier for a persistent RocksDB store so existing PLAIN data upgrades in place. In-memory and user-supplied stores keep their supplier and are upgraded on restore by the list-aware RecordConverter. The DslStoreSuppliers request stays PLAIN because the value-with-headers stores put one headers section in front of the whole value, which is not the shape of a list. - KStreamKStreamJoin / OuterJoinStoreWrapper: read the per-element headers back out of the store and emit them on the record produced for a non-joined record, so headers survive the outer-join delay rather than being dropped. This is the only PR of the three that changes behaviour: it is the switch that makes the new store reachable. It cannot compile without PR 1. Co-Authored-By: Claude Opus 5 (1M context) --- ...uterJoinListValueStoreRestorationTest.java | 164 +++++++++++++++ .../kstream/internals/KStreamKStreamJoin.java | 51 +++-- .../OuterStreamJoinStoreFactory.java | 50 +++-- .../internals/OuterJoinStoreWrapper.java | 135 +++++++++++++ .../KStreamKStreamOuterJoinTest.java | 191 ++++++++++++++++-- 5 files changed, 541 insertions(+), 50 deletions(-) create mode 100644 streams/src/main/java/org/apache/kafka/streams/state/internals/OuterJoinStoreWrapper.java diff --git a/streams/integration-tests/src/test/java/org/apache/kafka/streams/integration/OuterJoinListValueStoreRestorationTest.java b/streams/integration-tests/src/test/java/org/apache/kafka/streams/integration/OuterJoinListValueStoreRestorationTest.java index bdf824290bf32..83b5208f33530 100644 --- a/streams/integration-tests/src/test/java/org/apache/kafka/streams/integration/OuterJoinListValueStoreRestorationTest.java +++ b/streams/integration-tests/src/test/java/org/apache/kafka/streams/integration/OuterJoinListValueStoreRestorationTest.java @@ -17,7 +17,11 @@ package org.apache.kafka.streams.integration; import org.apache.kafka.clients.consumer.ConsumerConfig; +import org.apache.kafka.clients.consumer.ConsumerRecord; import org.apache.kafka.clients.producer.ProducerConfig; +import org.apache.kafka.common.header.Header; +import org.apache.kafka.common.header.Headers; +import org.apache.kafka.common.header.internals.RecordHeaders; import org.apache.kafka.common.serialization.Serdes; import org.apache.kafka.common.serialization.StringDeserializer; import org.apache.kafka.common.serialization.StringSerializer; @@ -42,9 +46,12 @@ import org.junit.jupiter.params.ParameterizedTest; import org.junit.jupiter.params.provider.Arguments; import org.junit.jupiter.params.provider.MethodSource; +import org.junit.jupiter.params.provider.ValueSource; +import java.nio.charset.StandardCharsets; import java.time.Duration; import java.util.List; +import java.util.Map; import java.util.Properties; import java.util.Set; import java.util.stream.Collectors; @@ -55,6 +62,7 @@ import static org.apache.kafka.streams.integration.utils.IntegrationTestUtils.startApplicationAndWaitUntilRunning; import static org.apache.kafka.streams.integration.utils.IntegrationTestUtils.waitForCompletion; import static org.apache.kafka.streams.integration.utils.IntegrationTestUtils.waitUntilMinKeyValueRecordsReceived; +import static org.apache.kafka.streams.integration.utils.IntegrationTestUtils.waitUntilMinRecordsReceived; import static org.apache.kafka.streams.utils.TestUtils.safeUniqueTestName; import static org.junit.jupiter.api.Assertions.assertEquals; @@ -228,6 +236,162 @@ public void testOuterJoinRestorationWithMultipleRecords(final String processingG "No unexpected keys should appear on the output topic after restoration"); } + /** + * Verifies that per-record headers stored in the header-aware outer-join store survive a full + * fault-tolerance cycle: the headers are written into the changelog, the local state is wiped, + * and after the store is rebuilt from the changelog the non-joined records are emitted carrying + * their original headers. + */ + @ParameterizedTest + @ValueSource(strings = {StreamsConfig.AT_LEAST_ONCE, StreamsConfig.EXACTLY_ONCE_V2}) + public void testOuterJoinHeadersSurviveRestoration(final String processingGuarantee) throws Exception { + // Headers are only stored in the header-aware store format. + streamsConfig.put(StreamsConfig.PROCESSING_GUARANTEE_CONFIG, processingGuarantee); + streamsConfig.put(StreamsConfig.DSL_STORE_FORMAT_CONFIG, StreamsConfig.DSL_STORE_FORMAT_HEADERS); + + // Step 1: Initial Topology Start + streams = createOuterJoinTopology(); + startApplicationAndWaitUntilRunning(streams); + + // Step 2: Produce non-joined left records, each carrying a distinct header. + // CRITICAL: Do NOT advance the window yet, so the records (and their headers) stay in the store. + final Map expectedHeaders = IntStream.range(0, 10).boxed() + .collect(Collectors.toMap(i -> "key" + i, i -> "v" + i)); + long timestamp = 1000L; + for (int i = 0; i < 10; i++) { + final Headers headers = new RecordHeaders().add("h", ("v" + i).getBytes(StandardCharsets.UTF_8)); + produceRecordWithHeaders(leftTopic, "key" + i, "left-" + i, headers, timestamp); + timestamp += 100; + } + + // Wait for processing and commit to the changelog, proven via a probe join result. + produceRecord(leftTopic, "probe", "probe-left", timestamp); + produceRecord(rightTopic, "probe", "probe-right", timestamp); + waitUntilMinKeyValueRecordsReceived(getConsumerConfig(), outputTopic, 1, 30000); + waitForCompletion(streams, 2, 30000); + + // Step 3: Force State Restoration (wipe local state, rebuild from changelog). + streams.close(Duration.ofSeconds(30)); + purgeLocalStreamsState(streamsConfig); + + // Step 4: Restart with Restoration + streams = createOuterJoinTopology(); + startApplicationAndWaitUntilRunning(streams); + + // Step 5: Advance the window to trigger emitNonJoinedOuterRecords() for the restored records. + produceRecord(leftTopic, "trigger", "trigger-value", 62000L); // beyond the 60-second window + + final List> results = + waitUntilMinRecordsReceived(getConsumerConfig(), outputTopic, 10, 30000); + + // Step 6: Each restored non-joined left record must be emitted with its original header. + final Map actualHeaders = results.stream() + .filter(r -> r.value() != null && r.value().endsWith("right=null")) + .filter(r -> r.key().startsWith("key")) + .collect(Collectors.toMap(ConsumerRecord::key, r -> headerValue(r, "h"), (a, b) -> a)); + + assertEquals(expectedHeaders, actualHeaders, + "Each non-joined left record should retain its original header after wipe + changelog restoration"); + } + + /** + * The downgrade direction: state is written by a HEADERS-format store and then restored by a PLAIN + * one. This is what the changelog encoding exists for. If the local + * {@code [headersSize][headers][flag][value]} element format reached the changelog, the PLAIN reader + * would consume each element's leading empty-headers {@code 0x00} as the {@code LeftOrRightValue} + * flag and silently emit left records as right ones -- i.e. {@code (null, x)} instead of + * {@code (x, null)} -- with no exception to signal it. + *

+ * Note this needs no version downgrade: flipping {@code dsl.store.format} back to PLAIN on one + * version is enough, and it is also the path the "delete the local state and rebuild the store from + * the changelog" advice in the {@code RocksDBStore} downgrade guard takes. + */ + @ParameterizedTest + @ValueSource(strings = {StreamsConfig.AT_LEAST_ONCE, StreamsConfig.EXACTLY_ONCE_V2}) + public void testHeadersStoreChangelogIsReadableByAPlainStore(final String processingGuarantee) throws Exception { + streamsConfig.put(StreamsConfig.PROCESSING_GUARANTEE_CONFIG, processingGuarantee); + streamsConfig.put(StreamsConfig.DSL_STORE_FORMAT_CONFIG, StreamsConfig.DSL_STORE_FORMAT_HEADERS); + + // Step 1: run in HEADERS mode and park non-joined left records (with headers) in the store. + streams = createOuterJoinTopology(); + startApplicationAndWaitUntilRunning(streams); + + long timestamp = 1000L; + for (int i = 0; i < 10; i++) { + final Headers headers = new RecordHeaders().add("h", ("v" + i).getBytes(StandardCharsets.UTF_8)); + produceRecordWithHeaders(leftTopic, "key" + i, "left-" + i, headers, timestamp); + timestamp += 100; + } + + produceRecord(leftTopic, "probe", "probe-left", timestamp); + produceRecord(rightTopic, "probe", "probe-right", timestamp); + waitUntilMinKeyValueRecordsReceived(getConsumerConfig(), outputTopic, 1, 30000); + waitForCompletion(streams, 2, 30000); + + // Step 2: wipe the local state, so the only surviving copy is the changelog. + streams.close(Duration.ofSeconds(30)); + purgeLocalStreamsState(streamsConfig); + + // Step 3: come back as a PLAIN store -- the downgrade. + streamsConfig.put(StreamsConfig.DSL_STORE_FORMAT_CONFIG, StreamsConfig.DSL_STORE_FORMAT_DEFAULT); + streams = createOuterJoinTopology(); + startApplicationAndWaitUntilRunning(streams); + + // Step 4: advance the window to emit the restored non-joined records. + produceRecord(leftTopic, "trigger", "trigger-value", 62000L); + + final List> results = + waitUntilMinKeyValueRecordsReceived(getConsumerConfig(), outputTopic, 10, 30000); + + final Set expectedKeys = IntStream.range(0, 10) + .mapToObj(i -> "key" + i) + .collect(Collectors.toSet()); + + // The join side must survive: every restored record is a LEFT record, so it is emitted as + // "left=, right=null". Under the old changelog format these came back as right records. + final Set emittedAsLeft = results.stream() + .filter(kv -> kv.value != null && kv.value.endsWith("right=null")) + .map(kv -> kv.key) + .collect(Collectors.toSet()); + assertEquals(expectedKeys, emittedAsLeft, + "A PLAIN store restoring a HEADERS-written changelog must still see these as left records"); + + // The values must survive too: the old format left a stray NUL prepended to each value. + final Map emittedValues = results.stream() + .filter(kv -> kv.key.startsWith("key")) + .collect(Collectors.toMap(kv -> kv.key, kv -> kv.value, (a, b) -> a)); + final Map expectedValues = IntStream.range(0, 10).boxed() + .collect(Collectors.toMap(i -> "key" + i, i -> "left=left-" + i + ", right=null")); + assertEquals(expectedValues, emittedValues, + "Values must be byte-identical after the downgrade, with no stray prefix"); + } + + private void produceRecordWithHeaders(final String topic, + final String key, + final String value, + final Headers headers, + final long timestamp) { + final Properties producerConfig = new Properties(); + producerConfig.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, CLUSTER.bootstrapServers()); + producerConfig.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class); + producerConfig.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class); + producerConfig.put(ProducerConfig.ACKS_CONFIG, "all"); + + IntegrationTestUtils.produceKeyValuesSynchronouslyWithTimestamp( + topic, + List.of(new KeyValue<>(key, value)), + producerConfig, + headers, + timestamp, + false + ); + } + + private static String headerValue(final ConsumerRecord record, final String key) { + final Header header = record.headers().lastHeader(key); + return header == null ? null : new String(header.value(), StandardCharsets.UTF_8); + } + private void produceRecord(final String topic, final String key, final String value, final long timestamp) { final Properties producerConfig = new Properties(); producerConfig.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, CLUSTER.bootstrapServers()); diff --git a/streams/src/main/java/org/apache/kafka/streams/kstream/internals/KStreamKStreamJoin.java b/streams/src/main/java/org/apache/kafka/streams/kstream/internals/KStreamKStreamJoin.java index 03ac193a0e035..a244262c73912 100644 --- a/streams/src/main/java/org/apache/kafka/streams/kstream/internals/KStreamKStreamJoin.java +++ b/streams/src/main/java/org/apache/kafka/streams/kstream/internals/KStreamKStreamJoin.java @@ -30,13 +30,14 @@ import org.apache.kafka.streams.processor.internals.StoreFactory; import org.apache.kafka.streams.processor.internals.StoreFactory.FactoryWrappingStoreBuilder; import org.apache.kafka.streams.processor.internals.metrics.StreamsMetricsImpl; +import org.apache.kafka.streams.state.AggregationWithHeaders; import org.apache.kafka.streams.state.KeyValueIterator; -import org.apache.kafka.streams.state.KeyValueStore; import org.apache.kafka.streams.state.StoreBuilder; import org.apache.kafka.streams.state.TimestampedWindowStoreWithHeaders; import org.apache.kafka.streams.state.ValueTimestampHeaders; import org.apache.kafka.streams.state.WindowStoreIterator; import org.apache.kafka.streams.state.internals.LeftOrRightValue; +import org.apache.kafka.streams.state.internals.OuterJoinStoreWrapper; import org.apache.kafka.streams.state.internals.TimestampedKeyAndJoinSide; import org.slf4j.Logger; @@ -102,7 +103,8 @@ public Set> stores() { protected abstract class KStreamKStreamJoinProcessor extends ContextualProcessor { private TimestampedWindowStoreWithHeaders otherWindowStore; private Sensor droppedRecordsSensor; - private Optional, LeftOrRightValue>> outerJoinStore = Optional.empty(); + private Optional> outerJoinStoreWrapper = Optional.empty(); + private boolean isOuterJoinStoreHeadersAware; private InternalProcessorContext internalProcessorContext; private TimeTracker sharedTimeTracker; @@ -117,7 +119,9 @@ public void init(final ProcessorContext context) { sharedTimeTracker = sharedTimeTrackerSupplier.get(context.taskId()); if (enableSpuriousResultFix) { - outerJoinStore = outerJoinWindowStoreFactory.map(s -> context.getStateStore(s.storeName())); + outerJoinStoreWrapper = outerJoinWindowStoreFactory + .map(s -> new OuterJoinStoreWrapper<>(context, s)); + isOuterJoinStoreHeadersAware = outerJoinStoreWrapper.map(OuterJoinStoreWrapper::isHeadersStore).orElse(false); sharedTimeTracker.setEmitInterval( StreamsConfig.InternalConfig.getLong( @@ -145,7 +149,7 @@ public void process(final Record record) { // Emit all non-joined records which window has closed if (inputRecordTimestamp == sharedTimeTracker.streamTime) { - outerJoinStore.ifPresent(store -> emitNonJoinedOuterRecords(store, record)); + outerJoinStoreWrapper.ifPresent(wrapper -> emitNonJoinedOuterRecords(wrapper, record)); } final long timeFrom = Math.max(0L, inputRecordTimestamp - joinBeforeMs); @@ -176,7 +180,7 @@ public void process(final Record record) { // // This condition below allows us to process the out-of-order records without the need // to hold it in the temporary outer store - if (outerJoinStore.isEmpty() || timeTo < sharedTimeTracker.streamTime) { + if (outerJoinStoreWrapper.isEmpty() || timeTo < sharedTimeTracker.streamTime) { context().forward(record.withValue(joiner.apply(record.key(), record.value(), null))); } else { sharedTimeTracker.updatedMinTime(inputRecordTimestamp); @@ -196,7 +200,7 @@ public void process(final Record record) { protected abstract VOther otherValue(final LeftOrRightValue leftOrRightValue); - private void emitNonJoinedOuterRecords(final KeyValueStore, LeftOrRightValue> store, + private void emitNonJoinedOuterRecords(final OuterJoinStoreWrapper wrapper, final Record record) { // calling `store.all()` creates an iterator what is an expensive operation on RocksDB; @@ -221,13 +225,13 @@ private void emitNonJoinedOuterRecords(final KeyValueStore, LeftOrRightValue> it = store.all()) { + try (final KeyValueIterator, AggregationWithHeaders>> it = wrapper.all()) { TimestampedKeyAndJoinSide prevKey = null; boolean outerJoinLeftWindowOpen = false; boolean outerJoinRightWindowOpen = false; while (it.hasNext()) { - final KeyValue, LeftOrRightValue> nextKeyValue = it.next(); + final KeyValue, AggregationWithHeaders>> nextKeyValue = it.next(); final TimestampedKeyAndJoinSide timestampedKeyAndJoinSide = nextKeyValue.key; sharedTimeTracker.minTime = timestampedKeyAndJoinSide.timestamp(); if (outerJoinLeftWindowOpen && outerJoinRightWindowOpen) { @@ -248,8 +252,8 @@ private void emitNonJoinedOuterRecords(final KeyValueStore leftOrRightValue = nextKeyValue.value; - forwardNonJoinedOuterRecords(record, timestampedKeyAndJoinSide, leftOrRightValue); + final AggregationWithHeaders> outerValue = nextKeyValue.value; + forwardNonJoinedOuterRecords(record, timestampedKeyAndJoinSide, outerValue); if (prevKey != null && !prevKey.equals(timestampedKeyAndJoinSide)) { // blind-delete the previous key from the outer window store now it is emitted; @@ -257,29 +261,32 @@ private void emitNonJoinedOuterRecords(final KeyValueStore record, final TimestampedKeyAndJoinSide timestampedKeyAndJoinSide, - final LeftOrRightValue leftOrRightValue) { + final AggregationWithHeaders> outerValue) { final K key = timestampedKeyAndJoinSide.key(); final long timestamp = timestampedKeyAndJoinSide.timestamp(); - final VThis thisValue = thisValue(leftOrRightValue); - final VOther otherValue = otherValue(leftOrRightValue); + final LeftOrRightValue storedValue = outerValue.aggregation(); + final VThis thisValue = thisValue(storedValue); + final VOther otherValue = otherValue(storedValue); final VOut nullJoinedValue = joiner.apply(key, thisValue, otherValue); - context().forward( - record.withKey(key).withValue(nullJoinedValue).withTimestamp(timestamp) - ); + Record forwarded = record.withKey(key).withValue(nullJoinedValue).withTimestamp(timestamp); + if (isOuterJoinStoreHeadersAware) { + forwarded = forwarded.withHeaders(outerValue.headers()); + } + context().forward(forwarded); } private boolean isOuterJoinWindowOpen(final TimestampedKeyAndJoinSide timestampedKeyAndJoinSide) { @@ -299,14 +306,14 @@ private long getOuterJoinLookBackTimeMs( private void emitInnerJoin(final Record thisRecord, final KeyValue otherRecord, final long inputRecordTimestamp) { - outerJoinStore.ifPresent(store -> { + outerJoinStoreWrapper.ifPresent(wrapper -> { // use putIfAbsent to first read and see if there's any values for the key, // if yes delete the key, otherwise do not issue a put; // we may delete some values with the same key early but since we are going // range over all values of the same key even after failure, since the other window-store // is only cleaned up by stream time, so this is okay for at-least-once. final TimestampedKeyAndJoinSide otherKey = makeOtherKey(thisRecord.key(), otherRecord.key); - store.putIfAbsent(otherKey, null); + wrapper.putIfAbsent(otherKey, null, null); }); context().forward( @@ -315,10 +322,10 @@ private void emitInnerJoin(final Record thisRecord, final KeyValue thisRecord) { - outerJoinStore.ifPresent(store -> { + outerJoinStoreWrapper.ifPresent(wrapper -> { final TimestampedKeyAndJoinSide thisKey = makeThisKey(thisRecord.key(), thisRecord.timestamp()); final LeftOrRightValue thisValue = makeThisValue(thisRecord.value()); - store.put(thisKey, thisValue); + wrapper.put(thisKey, thisValue, thisRecord.headers()); }); } diff --git a/streams/src/main/java/org/apache/kafka/streams/kstream/internals/OuterStreamJoinStoreFactory.java b/streams/src/main/java/org/apache/kafka/streams/kstream/internals/OuterStreamJoinStoreFactory.java index 0db5a36734c57..be73af6b9484d 100644 --- a/streams/src/main/java/org/apache/kafka/streams/kstream/internals/OuterStreamJoinStoreFactory.java +++ b/streams/src/main/java/org/apache/kafka/streams/kstream/internals/OuterStreamJoinStoreFactory.java @@ -23,15 +23,15 @@ import org.apache.kafka.streams.state.DslKeyValueParams; import org.apache.kafka.streams.state.DslStoreSuppliers; import org.apache.kafka.streams.state.KeyValueBytesStoreSupplier; -import org.apache.kafka.streams.state.KeyValueStore; import org.apache.kafka.streams.state.StoreBuilder; import org.apache.kafka.streams.state.Stores; +import org.apache.kafka.streams.state.internals.AggregationWithHeadersSerde; import org.apache.kafka.streams.state.internals.InMemoryWindowBytesStoreSupplier; -import org.apache.kafka.streams.state.internals.LeftOrRightValue; import org.apache.kafka.streams.state.internals.LeftOrRightValueSerde; import org.apache.kafka.streams.state.internals.ListValueStoreBuilder; +import org.apache.kafka.streams.state.internals.RocksDBKeyValueBytesStoreSupplier; +import org.apache.kafka.streams.state.internals.RocksDBListValueHeadersBytesStoreSupplier; import org.apache.kafka.streams.state.internals.RocksDbWindowBytesStoreSupplier; -import org.apache.kafka.streams.state.internals.TimestampedKeyAndJoinSide; import org.apache.kafka.streams.state.internals.TimestampedKeyAndJoinSideSerde; import java.time.Duration; @@ -96,7 +96,16 @@ public StoreBuilder builder() { final TimestampedKeyAndJoinSideSerde timestampedKeyAndJoinSideSerde = new TimestampedKeyAndJoinSideSerde<>(streamJoined.keySerde()); final LeftOrRightValueSerde leftOrRightValueSerde = new LeftOrRightValueSerde<>(streamJoined.valueSerde(), streamJoined.otherValueSerde()); - // Once the headers-aware version of ListValueStore is implemented (planned for AK 4.4), replace the PLAIN constant with the dslStoreFormat() method. + // We always ask the DslStoreSuppliers for a PLAIN bytes store, even in HEADERS mode. The + // value-with-headers stores put one headers section in front of the whole value, which is not + // the shape of a list: here each list element carries its own headers inline, so the generic + // whole-value converter would corrupt the blob. HEADERS mode instead swaps in the list-aware + // dual-column-family supplier below. + // + // This is a choice about the *local* store only. The changelog keeps the PLAIN element format + // regardless of mode -- see ChangeLoggingListValueBytesStoreWithHeaders, which strips the + // per-element headers into a record header before logging -- so an older version, or the same + // version with dsl.store.format flipped back to PLAIN, can still read the changelog. final DslKeyValueParams dslKeyValueParams = new DslKeyValueParams(name, DslStoreFormat.PLAIN); final KeyValueBytesStoreSupplier supplier; @@ -121,14 +130,31 @@ public StoreBuilder builder() { supplier = dslStoreSuppliers().keyValueStore(dslKeyValueParams); } - final StoreBuilder, LeftOrRightValue>> - builder = - new ListValueStoreBuilder<>( - supplier, - timestampedKeyAndJoinSideSerde, - leftOrRightValueSerde, - Time.SYSTEM - ); + final StoreBuilder builder; + if (dslStoreFormat() == DslStoreFormat.HEADERS) { + // For a persistent RocksDB store, use the dual-column-family variant so an existing store + // that was written by a pre-headers (PLAIN) version can be upgraded in place without + // corrupting old data. In-memory and user-supplied stores keep their supplier; their + // upgrade is handled on restore by the list-aware RecordConverter. + final KeyValueBytesStoreSupplier headersSupplier = + supplier instanceof RocksDBKeyValueBytesStoreSupplier + ? new RocksDBListValueHeadersBytesStoreSupplier(name) + : supplier; + builder = new ListValueStoreBuilder<>( + headersSupplier, + timestampedKeyAndJoinSideSerde, + new AggregationWithHeadersSerde<>(leftOrRightValueSerde), + Time.SYSTEM, + true + ); + } else { + builder = new ListValueStoreBuilder<>( + supplier, + timestampedKeyAndJoinSideSerde, + leftOrRightValueSerde, + Time.SYSTEM + ); + } if (loggingEnabled) { builder.withLoggingEnabled(streamJoined.logConfig()); diff --git a/streams/src/main/java/org/apache/kafka/streams/state/internals/OuterJoinStoreWrapper.java b/streams/src/main/java/org/apache/kafka/streams/state/internals/OuterJoinStoreWrapper.java new file mode 100644 index 0000000000000..e0afeefd68ec2 --- /dev/null +++ b/streams/src/main/java/org/apache/kafka/streams/state/internals/OuterJoinStoreWrapper.java @@ -0,0 +1,135 @@ +/* + * 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.header.Headers; +import org.apache.kafka.common.header.internals.RecordHeaders; +import org.apache.kafka.streams.DslStoreFormat; +import org.apache.kafka.streams.KeyValue; +import org.apache.kafka.streams.kstream.internals.AbstractConfigurableStoreFactory; +import org.apache.kafka.streams.processor.api.ProcessorContext; +import org.apache.kafka.streams.processor.internals.StoreFactory; +import org.apache.kafka.streams.state.AggregationWithHeaders; +import org.apache.kafka.streams.state.KeyValueIterator; +import org.apache.kafka.streams.state.KeyValueStore; + +/** + * Wraps the outer-join store used by {@code KStreamKStreamJoin} so the processor only deals + * with a single value shape: {@code AggregationWithHeaders>}. + *

+ * The value carries no timestamp: the entry timestamp is part of the key + * ({@link TimestampedKeyAndJoinSide#timestamp()}), which is what the entries are sorted by, so a + * separate value-side timestamp would be redundant. + *

+ * The underlying store is one of: + *

    + *
  • plain: {@code KeyValueStore, LeftOrRightValue>}
    • + *
  • headers-aware: {@code KeyValueStore, AggregationWithHeaders>>}
    • + *
+ * Both variants are wrapped by the same {@link MeteredKeyValueStore} class — they only differ + * in the value serde and (erased) generic parameter — so the variant cannot be detected by + * casting the runtime store. Instead, the variant is read from the {@link StoreFactory}'s + * configured {@link DslStoreFormat}. + */ +public class OuterJoinStoreWrapper { + + private final boolean isHeadersStore; + private KeyValueStore, LeftOrRightValue> plainStore; + private KeyValueStore, AggregationWithHeaders>> headersStore; + + public OuterJoinStoreWrapper(final ProcessorContext context, final StoreFactory storeFactory) { + this.isHeadersStore = isHeadersAware(storeFactory); + if (isHeadersStore) { + headersStore = context.getStateStore(storeFactory.storeName()); + } else { + plainStore = context.getStateStore(storeFactory.storeName()); + } + } + + private static boolean isHeadersAware(final StoreFactory storeFactory) { + if (storeFactory instanceof AbstractConfigurableStoreFactory) { + return ((AbstractConfigurableStoreFactory) storeFactory).dslStoreFormat() == DslStoreFormat.HEADERS; + } + return false; + } + + public boolean isHeadersStore() { + return isHeadersStore; + } + + public void put(final TimestampedKeyAndJoinSide key, + final LeftOrRightValue value, + final Headers headers) { + if (headersStore != null) { + headersStore.put(key, AggregationWithHeaders.make(value, headers)); + } else { + plainStore.put(key, value); + } + } + + public void putIfAbsent(final TimestampedKeyAndJoinSide key, + final LeftOrRightValue value, + final Headers headers) { + if (headersStore != null) { + headersStore.putIfAbsent(key, AggregationWithHeaders.make(value, headers)); + } else { + plainStore.putIfAbsent(key, value); + } + } + + public KeyValueIterator, AggregationWithHeaders>> all() { + if (headersStore != null) { + return headersStore.all(); + } + // The plain store has no per-element headers. Callers that need the timestamp read it from + // the key (TimestampedKeyAndJoinSide#timestamp). Callers that need the headers gate on + // isHeadersStore() — the lifted headers slot here is unused on the plain path. + return new LiftingIterator<>(plainStore.all()); + } + + private static final class LiftingIterator + implements KeyValueIterator, AggregationWithHeaders> { + + private final KeyValueIterator, V> inner; + + LiftingIterator(final KeyValueIterator, V> inner) { + this.inner = inner; + } + + @Override + public void close() { + inner.close(); + } + + @Override + public TimestampedKeyAndJoinSide peekNextKey() { + return inner.peekNextKey(); + } + + @Override + public boolean hasNext() { + return inner.hasNext(); + } + + @Override + public KeyValue, AggregationWithHeaders> next() { + final KeyValue, V> kv = inner.next(); + final AggregationWithHeaders lifted = AggregationWithHeaders.make(kv.value, new RecordHeaders()); + return KeyValue.pair(kv.key, lifted); + } + } +} diff --git a/streams/src/test/java/org/apache/kafka/streams/kstream/internals/KStreamKStreamOuterJoinTest.java b/streams/src/test/java/org/apache/kafka/streams/kstream/internals/KStreamKStreamOuterJoinTest.java index dd17dab0c662e..ef1880742885a 100644 --- a/streams/src/test/java/org/apache/kafka/streams/kstream/internals/KStreamKStreamOuterJoinTest.java +++ b/streams/src/test/java/org/apache/kafka/streams/kstream/internals/KStreamKStreamOuterJoinTest.java @@ -1074,22 +1074,41 @@ public void testShouldForwardCurrentHeaders(final boolean withHeaders) { (long) 211 )); - // Again, header forwarding is undefined, but the current observed behavior is that - // the headers pass through the forwarding record. - processor.checkAndClearProcessedRecords( - new Record<>( - 1, - "null+10", - 0L, - new RecordHeaders(new Header[]{new RecordHeader("h", new byte[]{0x3})}) - ), - new Record<>( - 0, - "A0+null", - 0L, - new RecordHeaders(new Header[]{new RecordHeader("h", new byte[]{0x3})}) - ) - ); + // Header forwarding for non-joined-outer emits depends on the store format: + // - HEADERS: the original outer record's headers are preserved + // - PLAIN: no per-record headers are stored, so the trigger record's headers + // pass through unchanged (legacy behavior). + if (withHeaders) { + processor.checkAndClearProcessedRecords( + new Record<>( + 1, + "null+10", + 0L, + new RecordHeaders(new Header[]{new RecordHeader("h", new byte[]{0x2})}) + ), + new Record<>( + 0, + "A0+null", + 0L, + new RecordHeaders(new Header[]{new RecordHeader("h", new byte[]{0x1})}) + ) + ); + } else { + processor.checkAndClearProcessedRecords( + new Record<>( + 1, + "null+10", + 0L, + new RecordHeaders(new Header[]{new RecordHeader("h", new byte[]{0x3})}) + ), + new Record<>( + 0, + "A0+null", + 0L, + new RecordHeaders(new Header[]{new RecordHeader("h", new byte[]{0x3})}) + ) + ); + } // verifies joined duplicates are emitted inputTopic1.pipeInput(new TestRecord<>( @@ -1116,6 +1135,146 @@ public void testShouldForwardCurrentHeaders(final boolean withHeaders) { } } + /** + * Header forwarding for non-joined-outer emits, multi-record case. The single-record version + * is in {@link #testShouldForwardCurrentHeaders}. This variant flushes several buffered + * outer records in one shot — it's the case where the headers-aware list store would expose + * "headers got mixed up between buffered entries" bugs (e.g. shared/aliased buffers, off-by-one + * deserialization), which the single-record test cannot catch. + */ + @ParameterizedTest + @ValueSource(booleans = {false, true}) + public void testShouldForwardPerRecordHeadersForMultipleBufferedOuterEmits(final boolean withHeaders) { + setDslStoreFormat(withHeaders); + final StreamsBuilder builder = new StreamsBuilder(); + + final KStream stream1; + final KStream stream2; + final KStream joined; + final MockApiProcessorSupplier supplier = new MockApiProcessorSupplier<>(); + stream1 = builder.stream(topic1, consumed); + stream2 = builder.stream(topic2, consumed2); + + joined = stream1.outerJoin( + stream2, + MockValueJoiner.TOSTRING_JOINER, + JoinWindows.ofTimeDifferenceAndGrace(ofMillis(100L), ofMillis(10L)), + StreamJoined.with(Serdes.Integer(), Serdes.String(), Serdes.Long()) + ); + joined.process(supplier); + + try (final TopologyTestDriver driver = new TopologyTestDriver(builder.build(), PROPS)) { + final TestInputTopic inputTopic1 = + driver.createInputTopic(topic1, new IntegerSerializer(), new StringSerializer(), Instant.ofEpochMilli(0L), Duration.ZERO); + final TestInputTopic inputTopic2 = + driver.createInputTopic(topic2, new IntegerSerializer(), new LongSerializer(), Instant.ofEpochMilli(0L), Duration.ZERO); + final MockApiProcessor processor = supplier.theCapturedProcessor(); + + // Four unmatched outer records, each at a distinct timestamp and carrying a distinct + // header byte. Timestamps are spread across the window so the buffered iterator + // returns them in a deterministic timestamp-ascending order. + inputTopic1.pipeInput(new TestRecord<>( + 10, + "A10", + new RecordHeaders(new Header[]{new RecordHeader("h", new byte[]{0x1})}), + 0L + )); + inputTopic2.pipeInput(new TestRecord<>( + 11, + 20L, + new RecordHeaders(new Header[]{new RecordHeader("h", new byte[]{0x2})}), + 1L + )); + inputTopic1.pipeInput(new TestRecord<>( + 12, + "A12", + new RecordHeaders(new Header[]{new RecordHeader("h", new byte[]{0x3})}), + 2L + )); + inputTopic2.pipeInput(new TestRecord<>( + 13, + 30L, + new RecordHeaders(new Header[]{new RecordHeader("h", new byte[]{0x4})}), + 3L + )); + + // Sanity: nothing has been emitted yet — windows are still open. + processor.checkAndClearProcessedRecords(); + + // Trigger: bumps stream-time well past windowEnd + grace for all four buffered records, + // causing them to flush as null-side outer-join emits. The trigger itself is buffered + // (its own window has not closed) and must NOT appear in the emitted output. + inputTopic1.pipeInput(new TestRecord<>( + 99, + "TRIG", + new RecordHeaders(new Header[]{new RecordHeader("h", new byte[]{0x9})}), + 500L + )); + + if (withHeaders) { + // Each buffered record's own original header must come back out — no leakage from + // the trigger record (0x9), and no cross-contamination between buffered entries. + processor.checkAndClearProcessedRecords( + new Record<>( + 10, + "A10+null", + 0L, + new RecordHeaders(new Header[]{new RecordHeader("h", new byte[]{0x1})}) + ), + new Record<>( + 11, + "null+20", + 1L, + new RecordHeaders(new Header[]{new RecordHeader("h", new byte[]{0x2})}) + ), + new Record<>( + 12, + "A12+null", + 2L, + new RecordHeaders(new Header[]{new RecordHeader("h", new byte[]{0x3})}) + ), + new Record<>( + 13, + "null+30", + 3L, + new RecordHeaders(new Header[]{new RecordHeader("h", new byte[]{0x4})}) + ) + ); + } else { + // Legacy plain-store behavior: the buffer carries no headers, so every emit + // picks up the trigger record's headers (0x9). This is the documented "leak" + // behavior pinned by testShouldForwardCurrentHeaders, asserted here across + // multiple emits to catch any future regression that leaks a different value. + processor.checkAndClearProcessedRecords( + new Record<>( + 10, + "A10+null", + 0L, + new RecordHeaders(new Header[]{new RecordHeader("h", new byte[]{0x9})}) + ), + new Record<>( + 11, + "null+20", + 1L, + new RecordHeaders(new Header[]{new RecordHeader("h", new byte[]{0x9})}) + ), + new Record<>( + 12, + "A12+null", + 2L, + new RecordHeaders(new Header[]{new RecordHeader("h", new byte[]{0x9})}) + ), + new Record<>( + 13, + "null+30", + 3L, + new RecordHeaders(new Header[]{new RecordHeader("h", new byte[]{0x9})}) + ) + ); + } + } + } + private void testUpperWindowBound(final int[] expectedKeys, final TopologyTestDriver driver, final MockApiProcessor processor) { From 7e36db71c662f39ed197ce7a35c6026d3bcf53ba Mon Sep 17 00:00:00 2001 From: Alieh Saeedi Date: Tue, 28 Jul 2026 20:06:49 +0200 Subject: [PATCH 5/6] KAFKA-20413: Adapt the outer-join wiring to the reviewed [1/3] [1/3] dropped ListValueStoreBuilder's explicit headersFormat flag and now infers it from the supplier being a HeadersBytesStoreSupplier, so the 5-arg constructor this factory called is gone. Derive the element serde from the same signal, so HEADERS mode only takes effect when the supplier can actually provide a headers-aware bytes store. Pairing headers-format elements with the PLAIN changelogger would put each element's empty-headers 0x00 prefix on the changelog, where an old PLAIN reader takes it for the LeftOrRightValue flag. For the same reason OuterJoinStoreWrapper now reads the variant off the bytes store's HeadersAwareListValueStore marker instead of the configured DslStoreFormat, which would have claimed headers for an in-memory or user-supplied store that is in fact holding PLAIN elements. Co-Authored-By: Claude Opus 5 (1M context) --- .../OuterStreamJoinStoreFactory.java | 27 +++++++++------- .../internals/OuterJoinStoreWrapper.java | 31 ++++++++++--------- 2 files changed, 31 insertions(+), 27 deletions(-) diff --git a/streams/src/main/java/org/apache/kafka/streams/kstream/internals/OuterStreamJoinStoreFactory.java b/streams/src/main/java/org/apache/kafka/streams/kstream/internals/OuterStreamJoinStoreFactory.java index be73af6b9484d..697b2bcafa6ee 100644 --- a/streams/src/main/java/org/apache/kafka/streams/kstream/internals/OuterStreamJoinStoreFactory.java +++ b/streams/src/main/java/org/apache/kafka/streams/kstream/internals/OuterStreamJoinStoreFactory.java @@ -131,21 +131,24 @@ public StoreBuilder builder() { } final StoreBuilder builder; - if (dslStoreFormat() == DslStoreFormat.HEADERS) { - // For a persistent RocksDB store, use the dual-column-family variant so an existing store - // that was written by a pre-headers (PLAIN) version can be upgraded in place without - // corrupting old data. In-memory and user-supplied stores keep their supplier; their - // upgrade is handled on restore by the list-aware RecordConverter. - final KeyValueBytesStoreSupplier headersSupplier = - supplier instanceof RocksDBKeyValueBytesStoreSupplier - ? new RocksDBListValueHeadersBytesStoreSupplier(name) - : supplier; + // HEADERS mode needs a bytes store that can actually hold the headers element format: only the + // dual-column-family RocksDB variant can, so that a store already written by a pre-headers + // (PLAIN) version is upgraded in place rather than read as corrupt. In-memory and user-supplied + // suppliers stay on PLAIN elements, which is what ordinary key-value stores already do in + // HEADERS mode: InMemoryDslStoreSuppliers#keyValueStore ignores the format, and + // KeyValueStoreMaterializer picks its builder off the supplier rather than off the format. + if (dslStoreFormat() == DslStoreFormat.HEADERS && supplier instanceof RocksDBKeyValueBytesStoreSupplier) { + // All three decisions follow this one branch: the element serde here, the changelogger via + // ListValueStoreBuilder's HeadersBytesStoreSupplier check, and the processor-side value shape + // via OuterJoinStoreWrapper's HeadersAwareListValueStore check. They must not disagree -- + // headers-format elements written through the PLAIN changelogger would put each element's + // empty-headers 0x00 prefix on the changelog, where an old PLAIN reader takes it for the + // LeftOrRightValue flag and silently turns left values into right ones. builder = new ListValueStoreBuilder<>( - headersSupplier, + new RocksDBListValueHeadersBytesStoreSupplier(name), timestampedKeyAndJoinSideSerde, new AggregationWithHeadersSerde<>(leftOrRightValueSerde), - Time.SYSTEM, - true + Time.SYSTEM ); } else { builder = new ListValueStoreBuilder<>( diff --git a/streams/src/main/java/org/apache/kafka/streams/state/internals/OuterJoinStoreWrapper.java b/streams/src/main/java/org/apache/kafka/streams/state/internals/OuterJoinStoreWrapper.java index e0afeefd68ec2..c00d44d0ef342 100644 --- a/streams/src/main/java/org/apache/kafka/streams/state/internals/OuterJoinStoreWrapper.java +++ b/streams/src/main/java/org/apache/kafka/streams/state/internals/OuterJoinStoreWrapper.java @@ -20,7 +20,7 @@ import org.apache.kafka.common.header.internals.RecordHeaders; import org.apache.kafka.streams.DslStoreFormat; import org.apache.kafka.streams.KeyValue; -import org.apache.kafka.streams.kstream.internals.AbstractConfigurableStoreFactory; +import org.apache.kafka.streams.processor.StateStore; import org.apache.kafka.streams.processor.api.ProcessorContext; import org.apache.kafka.streams.processor.internals.StoreFactory; import org.apache.kafka.streams.state.AggregationWithHeaders; @@ -40,10 +40,16 @@ *
  • plain: {@code KeyValueStore, LeftOrRightValue>}
    • *
  • headers-aware: {@code KeyValueStore, AggregationWithHeaders>>}
    • * - * Both variants are wrapped by the same {@link MeteredKeyValueStore} class — they only differ - * in the value serde and (erased) generic parameter — so the variant cannot be detected by - * casting the runtime store. Instead, the variant is read from the {@link StoreFactory}'s - * configured {@link DslStoreFormat}. + * Both variants are wrapped by the same {@link MeteredKeyValueStore} class — they only differ in the + * value serde and (erased) generic parameter — so the variant cannot be read off the outermost store. + * It is read instead from the innermost bytes store, which carries the + * {@link HeadersAwareListValueStore} marker exactly when the elements are in the headers format. + *

    + * That is deliberately the same signal {@code OuterStreamJoinStoreFactory} uses to pick the + * element serde and {@link ListValueStoreBuilder} uses to pick the changelogger, rather than the + * configured {@link DslStoreFormat}: HEADERS only takes effect if the supplier could actually provide a + * headers-aware bytes store, so keying off the format would claim headers for an in-memory or + * user-supplied store that is in fact holding PLAIN elements. */ public class OuterJoinStoreWrapper { @@ -51,20 +57,15 @@ public class OuterJoinStoreWrapper { private KeyValueStore, LeftOrRightValue> plainStore; private KeyValueStore, AggregationWithHeaders>> headersStore; + @SuppressWarnings("unchecked") public OuterJoinStoreWrapper(final ProcessorContext context, final StoreFactory storeFactory) { - this.isHeadersStore = isHeadersAware(storeFactory); + final StateStore store = context.getStateStore(storeFactory.storeName()); + this.isHeadersStore = WrappedStateStore.isHeadersAwareListValue(store); if (isHeadersStore) { - headersStore = context.getStateStore(storeFactory.storeName()); + headersStore = (KeyValueStore, AggregationWithHeaders>>) store; } else { - plainStore = context.getStateStore(storeFactory.storeName()); - } - } - - private static boolean isHeadersAware(final StoreFactory storeFactory) { - if (storeFactory instanceof AbstractConfigurableStoreFactory) { - return ((AbstractConfigurableStoreFactory) storeFactory).dslStoreFormat() == DslStoreFormat.HEADERS; + plainStore = (KeyValueStore, LeftOrRightValue>) store; } - return false; } public boolean isHeadersStore() { From 399ea232c39910e8b603b2280c570234ec146e88 Mon Sep 17 00:00:00 2001 From: Alieh Saeedi Date: Tue, 28 Jul 2026 20:55:55 +0200 Subject: [PATCH 6/6] fix outer-join wiring --- ...uterJoinListValueStoreRestorationTest.java | 60 ++++++++++++++++++- .../kstream/internals/KStreamKStreamJoin.java | 24 ++++---- .../internals/OuterJoinStoreWrapper.java | 48 +++++++++++++-- .../KStreamKStreamOuterJoinTest.java | 2 +- 4 files changed, 115 insertions(+), 19 deletions(-) diff --git a/streams/integration-tests/src/test/java/org/apache/kafka/streams/integration/OuterJoinListValueStoreRestorationTest.java b/streams/integration-tests/src/test/java/org/apache/kafka/streams/integration/OuterJoinListValueStoreRestorationTest.java index 83b5208f33530..f0ce98524a7ce 100644 --- a/streams/integration-tests/src/test/java/org/apache/kafka/streams/integration/OuterJoinListValueStoreRestorationTest.java +++ b/streams/integration-tests/src/test/java/org/apache/kafka/streams/integration/OuterJoinListValueStoreRestorationTest.java @@ -19,9 +19,11 @@ import org.apache.kafka.clients.consumer.ConsumerConfig; import org.apache.kafka.clients.consumer.ConsumerRecord; import org.apache.kafka.clients.producer.ProducerConfig; +import org.apache.kafka.common.IsolationLevel; import org.apache.kafka.common.header.Header; import org.apache.kafka.common.header.Headers; import org.apache.kafka.common.header.internals.RecordHeaders; +import org.apache.kafka.common.serialization.ByteArrayDeserializer; import org.apache.kafka.common.serialization.Serdes; import org.apache.kafka.common.serialization.StringDeserializer; import org.apache.kafka.common.serialization.StringSerializer; @@ -52,6 +54,7 @@ import java.time.Duration; import java.util.List; import java.util.Map; +import java.util.Objects; import java.util.Properties; import java.util.Set; import java.util.stream.Collectors; @@ -65,6 +68,7 @@ import static org.apache.kafka.streams.integration.utils.IntegrationTestUtils.waitUntilMinRecordsReceived; import static org.apache.kafka.streams.utils.TestUtils.safeUniqueTestName; import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertTrue; /** * Integration test for verifying ListValueStore deserialization behavior after state restoration @@ -77,6 +81,16 @@ public class OuterJoinListValueStoreRestorationTest { private static final int NUM_BROKERS = 1; public static final EmbeddedKafkaCluster CLUSTER = new EmbeddedKafkaCluster(NUM_BROKERS); + /** Stands in for an absent record header, so a lost header shows up as a diff rather than an NPE. */ + private static final String NO_HEADER = ""; + + /** + * {@code ListValueStoreUpgradeUtils.LIST_VALUE_HEADERS_HEADER_KEY}, repeated as a literal because that + * constant is package-private in {@code org.apache.kafka.streams.state.internals}. Only the + * headers-format changelogger writes it, which makes it the marker that HEADERS mode really took hold. + */ + private static final String LIST_VALUE_HEADERS_HEADER_KEY = "vh"; + private String applicationId; private String leftTopic; private String rightTopic; @@ -285,10 +299,16 @@ public void testOuterJoinHeadersSurviveRestoration(final String processingGuaran waitUntilMinRecordsReceived(getConsumerConfig(), outputTopic, 10, 30000); // Step 6: Each restored non-joined left record must be emitted with its original header. + // A missing header maps to a sentinel rather than null: that is the regression this test guards, + // and Collectors.toMap rejects null values, so it would surface as a bare NPE here instead of an + // assertEquals diff showing which keys lost their header. final Map actualHeaders = results.stream() .filter(r -> r.value() != null && r.value().endsWith("right=null")) .filter(r -> r.key().startsWith("key")) - .collect(Collectors.toMap(ConsumerRecord::key, r -> headerValue(r, "h"), (a, b) -> a)); + .collect(Collectors.toMap( + ConsumerRecord::key, + r -> Objects.requireNonNullElse(headerValue(r, "h"), NO_HEADER), + (a, b) -> a)); assertEquals(expectedHeaders, actualHeaders, "Each non-joined left record should retain its original header after wipe + changelog restoration"); @@ -328,6 +348,15 @@ public void testHeadersStoreChangelogIsReadableByAPlainStore(final String proces waitUntilMinKeyValueRecordsReceived(getConsumerConfig(), outputTopic, 1, 30000); waitForCompletion(streams, 2, 30000); + // Step 1b: prove HEADERS mode actually took hold before relying on the downgrade below. Without + // this the test is vacuous: if the outer-join store ignored dsl.store.format, step 1 would log + // PLAIN elements and step 3 would read them back happily, so every assertion below would pass for + // the wrong reason -- including with this whole feature reverted. + assertTrue( + outerJoinChangelogCarriesElementHeaders(), + "No changelog record carried the " + LIST_VALUE_HEADERS_HEADER_KEY + " header, so the store was " + + "not the headers-format one and there is nothing for the downgrade below to prove"); + // Step 2: wipe the local state, so the only surviving copy is the changelog. streams.close(Duration.ofSeconds(30)); purgeLocalStreamsState(streamsConfig); @@ -387,6 +416,35 @@ private void produceRecordWithHeaders(final String topic, ); } + /** + * Whether the outer-join store's changelog carries the per-element headers control header, i.e. + * whether the store writing it was the headers-format one. + */ + private boolean outerJoinChangelogCarriesElementHeaders() throws Exception { + final Set topics = CLUSTER.getAllTopicsInCluster(); + // The topology does not name the join store, so it is KSTREAM-OUTERSHARED--store. + final String changelogTopic = topics.stream() + .filter(t -> t.contains("OUTERSHARED") && t.endsWith("-store-changelog")) + .findFirst() + .orElseThrow(() -> new AssertionError("no outer-join changelog topic among " + topics)); + + final Properties consumerConfig = new Properties(); + consumerConfig.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, CLUSTER.bootstrapServers()); + consumerConfig.put(ConsumerConfig.GROUP_ID_CONFIG, "changelog-consumer-" + applicationId); + consumerConfig.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, ByteArrayDeserializer.class); + consumerConfig.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, ByteArrayDeserializer.class); + consumerConfig.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest"); + // Under exactly-once the changelog writes are transactional, so an uncommitted read could see + // records from an aborted transaction -- or miss the committed ones entirely. + consumerConfig.put(ConsumerConfig.ISOLATION_LEVEL_CONFIG, IsolationLevel.READ_COMMITTED.toString()); + + final List> changelogRecords = + waitUntilMinRecordsReceived(consumerConfig, changelogTopic, 1, 30000); + + return changelogRecords.stream() + .anyMatch(r -> r.headers().lastHeader(LIST_VALUE_HEADERS_HEADER_KEY) != null); + } + private static String headerValue(final ConsumerRecord record, final String key) { final Header header = record.headers().lastHeader(key); return header == null ? null : new String(header.value(), StandardCharsets.UTF_8); diff --git a/streams/src/main/java/org/apache/kafka/streams/kstream/internals/KStreamKStreamJoin.java b/streams/src/main/java/org/apache/kafka/streams/kstream/internals/KStreamKStreamJoin.java index a244262c73912..1e03fdca21c8d 100644 --- a/streams/src/main/java/org/apache/kafka/streams/kstream/internals/KStreamKStreamJoin.java +++ b/streams/src/main/java/org/apache/kafka/streams/kstream/internals/KStreamKStreamJoin.java @@ -104,7 +104,6 @@ protected abstract class KStreamKStreamJoinProcessor extends ContextualProcessor private TimestampedWindowStoreWithHeaders otherWindowStore; private Sensor droppedRecordsSensor; private Optional> outerJoinStoreWrapper = Optional.empty(); - private boolean isOuterJoinStoreHeadersAware; private InternalProcessorContext internalProcessorContext; private TimeTracker sharedTimeTracker; @@ -121,7 +120,6 @@ public void init(final ProcessorContext context) { if (enableSpuriousResultFix) { outerJoinStoreWrapper = outerJoinWindowStoreFactory .map(s -> new OuterJoinStoreWrapper<>(context, s)); - isOuterJoinStoreHeadersAware = outerJoinStoreWrapper.map(OuterJoinStoreWrapper::isHeadersStore).orElse(false); sharedTimeTracker.setEmitInterval( StreamsConfig.InternalConfig.getLong( @@ -253,27 +251,29 @@ private void emitNonJoinedOuterRecords(final OuterJoinStoreWrapper> outerValue = nextKeyValue.value; - forwardNonJoinedOuterRecords(record, timestampedKeyAndJoinSide, outerValue); + forwardNonJoinedOuterRecords(wrapper, record, timestampedKeyAndJoinSide, outerValue); if (prevKey != null && !prevKey.equals(timestampedKeyAndJoinSide)) { // blind-delete the previous key from the outer window store now it is emitted; // we do this because this delete would remove the whole list of values of the same key, // and hence if we delete eagerly and then fail, we would miss emitting join results of the later // values in the list. - // we do not use delete() calls since it would incur extra get() - wrapper.put(prevKey, null, null); + // this is a blind write of a whole-list tombstone, not the store's delete(), which + // would incur an extra get() to return the previous value we have no use for + wrapper.deleteList(prevKey); } prevKey = timestampedKeyAndJoinSide; } // at the end of the iteration, we need to delete the last key if (prevKey != null) { - wrapper.put(prevKey, null, null); + wrapper.deleteList(prevKey); } } } - private void forwardNonJoinedOuterRecords(final Record record, + private void forwardNonJoinedOuterRecords(final OuterJoinStoreWrapper wrapper, + final Record record, final TimestampedKeyAndJoinSide timestampedKeyAndJoinSide, final AggregationWithHeaders> outerValue) { final K key = timestampedKeyAndJoinSide.key(); @@ -283,7 +283,9 @@ private void forwardNonJoinedOuterRecords(final Record record, final VOther otherValue = otherValue(storedValue); final VOut nullJoinedValue = joiner.apply(key, thisValue, otherValue); Record forwarded = record.withKey(key).withValue(nullJoinedValue).withTimestamp(timestamp); - if (isOuterJoinStoreHeadersAware) { + // Only a headers store actually kept the buffered record's headers; on the plain path the + // lifted headers are empty and the record keeps whatever it was forwarded with. + if (wrapper.isHeadersStore()) { forwarded = forwarded.withHeaders(outerValue.headers()); } context().forward(forwarded); @@ -307,13 +309,13 @@ private long getOuterJoinLookBackTimeMs( private void emitInnerJoin(final Record thisRecord, final KeyValue otherRecord, final long inputRecordTimestamp) { outerJoinStoreWrapper.ifPresent(wrapper -> { - // use putIfAbsent to first read and see if there's any values for the key, - // if yes delete the key, otherwise do not issue a put; + // this first reads to see if there are any values for the key, and only then deletes + // them, so that no tombstone is written for a key that was never buffered; // we may delete some values with the same key early but since we are going // range over all values of the same key even after failure, since the other window-store // is only cleaned up by stream time, so this is okay for at-least-once. final TimestampedKeyAndJoinSide otherKey = makeOtherKey(thisRecord.key(), otherRecord.key); - wrapper.putIfAbsent(otherKey, null, null); + wrapper.deleteListIfPresent(otherKey); }); context().forward( diff --git a/streams/src/main/java/org/apache/kafka/streams/state/internals/OuterJoinStoreWrapper.java b/streams/src/main/java/org/apache/kafka/streams/state/internals/OuterJoinStoreWrapper.java index c00d44d0ef342..9df2891849c12 100644 --- a/streams/src/main/java/org/apache/kafka/streams/state/internals/OuterJoinStoreWrapper.java +++ b/streams/src/main/java/org/apache/kafka/streams/state/internals/OuterJoinStoreWrapper.java @@ -27,6 +27,8 @@ import org.apache.kafka.streams.state.KeyValueIterator; import org.apache.kafka.streams.state.KeyValueStore; +import java.util.Objects; + /** * Wraps the outer-join store used by {@code KStreamKStreamJoin} so the processor only deals * with a single value shape: {@code AggregationWithHeaders>}. @@ -53,6 +55,17 @@ */ public class OuterJoinStoreWrapper { + /** + * The headers every entry on the plain path is lifted with. Shared rather than allocated per entry: + * the flush loop never reads them back (it gates on {@link #isHeadersStore()}), and read-only means + * that a future caller which does read them cannot mutate what every other entry sees. + */ + private static final RecordHeaders EMPTY_HEADERS = new RecordHeaders(); + + static { + EMPTY_HEADERS.setReadOnly(); + } + private final boolean isHeadersStore; private KeyValueStore, LeftOrRightValue> plainStore; private KeyValueStore, AggregationWithHeaders>> headersStore; @@ -72,9 +85,17 @@ public boolean isHeadersStore() { return isHeadersStore; } + /** + * Appends {@code value} to the list held under {@code key}. Both arguments are required — the + * whole-list delete is {@link #deleteList(TimestampedKeyAndJoinSide)}, not a null value. + */ public void put(final TimestampedKeyAndJoinSide key, final LeftOrRightValue value, final Headers headers) { + Objects.requireNonNull(value, "value must not be null; use deleteList to remove the list"); + // AggregationWithHeaders rejects null headers, but only once the value is non-null, so without + // this the mistake would surface as an NPE from inside the value class instead of from here. + Objects.requireNonNull(headers, "headers must not be null"); if (headersStore != null) { headersStore.put(key, AggregationWithHeaders.make(value, headers)); } else { @@ -82,13 +103,28 @@ public void put(final TimestampedKeyAndJoinSide key, } } - public void putIfAbsent(final TimestampedKeyAndJoinSide key, - final LeftOrRightValue value, - final Headers headers) { + /** + * Removes every value held under {@code key}. Spelled as a delete rather than a null put because + * {@link ListValueStore#put(org.apache.kafka.common.utils.Bytes, byte[])} with a null value drops the + * key's whole list, not just one element. + */ + public void deleteList(final TimestampedKeyAndJoinSide key) { + if (headersStore != null) { + headersStore.put(key, null); + } else { + plainStore.put(key, null); + } + } + + /** + * Removes every value held under {@code key}, but only if the key has any — a no-op otherwise, so + * that no tombstone is written for a key that was never there. + */ + public void deleteListIfPresent(final TimestampedKeyAndJoinSide key) { if (headersStore != null) { - headersStore.putIfAbsent(key, AggregationWithHeaders.make(value, headers)); + headersStore.putIfAbsent(key, null); } else { - plainStore.putIfAbsent(key, value); + plainStore.putIfAbsent(key, null); } } @@ -129,7 +165,7 @@ public boolean hasNext() { @Override public KeyValue, AggregationWithHeaders> next() { final KeyValue, V> kv = inner.next(); - final AggregationWithHeaders lifted = AggregationWithHeaders.make(kv.value, new RecordHeaders()); + final AggregationWithHeaders lifted = AggregationWithHeaders.make(kv.value, EMPTY_HEADERS); return KeyValue.pair(kv.key, lifted); } } diff --git a/streams/src/test/java/org/apache/kafka/streams/kstream/internals/KStreamKStreamOuterJoinTest.java b/streams/src/test/java/org/apache/kafka/streams/kstream/internals/KStreamKStreamOuterJoinTest.java index ef1880742885a..59225b93f592c 100644 --- a/streams/src/test/java/org/apache/kafka/streams/kstream/internals/KStreamKStreamOuterJoinTest.java +++ b/streams/src/test/java/org/apache/kafka/streams/kstream/internals/KStreamKStreamOuterJoinTest.java @@ -1163,7 +1163,7 @@ public void testShouldForwardPerRecordHeadersForMultipleBufferedOuterEmits(final ); joined.process(supplier); - try (final TopologyTestDriver driver = new TopologyTestDriver(builder.build(), PROPS)) { + try (final TopologyTestDriver driver = new TopologyTestDriverBuilder(builder.build()).withConfig(PROPS).build()) { final TestInputTopic inputTopic1 = driver.createInputTopic(topic1, new IntegerSerializer(), new StringSerializer(), Instant.ofEpochMilli(0L), Duration.ZERO); final TestInputTopic inputTopic2 =