mjsax commented on code in PR #22961:
URL: https://github.com/apache/kafka/pull/22961#discussion_r3663472755
##########
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(),
Review Comment:
Do we want to keep the `record.headers()` unmodified? Should we at least
strip our own header `LIST_VALUE_HEADERS_HEADER_KEY` (would also be a side
effect inside `istValueStoreUpgradeUtils.elementHeaders(...)` ?
Or maybe it does not matter, as this `ConsumerRecord` should be short lived
anyway, so we don't really delay freeing up he memory for too long.
##########
streams/src/main/java/org/apache/kafka/streams/state/internals/ListValueStoreBuilder.java:
##########
@@ -26,15 +26,25 @@
public class ListValueStoreBuilder<K, V> extends AbstractStoreBuilder<K, V,
KeyValueStore<K, V>> {
private final KeyValueBytesStoreSupplier storeSupplier;
+ private final boolean headersFormat;
public ListValueStoreBuilder(final KeyValueBytesStoreSupplier
storeSupplier,
final Serde<K> keySerde,
final Serde<V> valueSerde,
final Time time) {
+ this(storeSupplier, keySerde, valueSerde, time, false);
+ }
+
+ public ListValueStoreBuilder(final KeyValueBytesStoreSupplier
storeSupplier,
+ final Serde<K> keySerde,
+ final Serde<V> 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;
Review Comment:
Do we need to pass this boolean explicitly, or should we try to inver from
the `KeyValueBytesStoreSupplier` (ie, some `instanceof <markerInterface>` check
-- should we use existing `HeadersBytesStoreSupplier` interface (which
`RocksDBListValueHeadersBytesStoreSupplier` would implement)?
##########
streams/src/main/java/org/apache/kafka/streams/state/internals/ListValueStoreUpgradeUtils.java:
##########
@@ -0,0 +1,238 @@
+/*
+ * 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.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;
+
+import java.io.ByteArrayOutputStream;
+import java.nio.ByteBuffer;
+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).
+ * <p>
+ * 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}:
+ * <ul>
+ * <li>PLAIN: {@code [leftFlag(1B)][rawValue]} (a {@code
LeftOrRightValue})</li>
+ * <li>HEADERS: {@code
[headersSize(varint)][headersBytes][leftFlag(1B)][rawValue]}
+ * (an {@code AggregationWithHeaders<LeftOrRightValue>})</li>
+ * </ul>
+ * A PLAIN element becomes a HEADERS element with <em>empty</em> 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}.
+ * <p>
+ * The HEADERS element format above is the <em>local, on-disk</em> 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 <em>value</em> can stay in the format an old
PLAIN store understands.
+ * <p>
+ * 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.
+ * <p>
+ * Deliberately namespaced to avoid colliding with user headers that ride
along on the record.
+ * A record <em>without</em> 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<byte[]>> LIST_SERDE =
Serdes.ListSerde(ArrayList.class, Serdes.ByteArray());
Review Comment:
It it must match, why not share it? Make one accessible by the user on just
use?
##########
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.
+ * <p>
+ * 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}):
+ * <ul>
+ * <li>DEFAULT: legacy PLAIN list blobs written by the pre-headers
version;</li>
+ * <li>{@code listValueWithHeaders}: list blobs whose elements carry inline
headers.</li>
+ * </ul>
+ * 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.
+ * <p>
+ * This class intentionally does NOT implement {@code HeadersBytesStore}; see
+ * {@link HeadersAwareListValueStore} for why.
Review Comment:
But should it implement `HeadersAwareListValueStore` instead?
##########
streams/src/main/java/org/apache/kafka/streams/state/internals/ListValueStoreUpgradeUtils.java:
##########
@@ -0,0 +1,238 @@
+/*
+ * 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.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;
+
+import java.io.ByteArrayOutputStream;
+import java.nio.ByteBuffer;
+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).
+ * <p>
+ * 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}:
+ * <ul>
+ * <li>PLAIN: {@code [leftFlag(1B)][rawValue]} (a {@code
LeftOrRightValue})</li>
+ * <li>HEADERS: {@code
[headersSize(varint)][headersBytes][leftFlag(1B)][rawValue]}
+ * (an {@code AggregationWithHeaders<LeftOrRightValue>})</li>
+ * </ul>
+ * A PLAIN element becomes a HEADERS element with <em>empty</em> 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}.
+ * <p>
+ * The HEADERS element format above is the <em>local, on-disk</em> 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 <em>value</em> can stay in the format an old
PLAIN store understands.
+ * <p>
+ * 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.
+ * <p>
+ * Deliberately namespaced to avoid colliding with user headers that ride
along on the record.
+ * A record <em>without</em> 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<byte[]>> 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<byte[]> plainElements =
LIST_SERDE.deserializer().deserialize(null, plainListBlob);
+ final List<byte[]> 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);
+ }
+
+ /**
+ * 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[])}.
+ * <p>
+ * 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<byte[]> headersElements =
LIST_SERDE.deserializer().deserialize(null, headersListBlob);
+ final List<byte[]> 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<byte[]> plainElements =
LIST_SERDE.deserializer().deserialize(null, plainListBlob);
+ final List<byte[]> 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) {
Review Comment:
I believe this case can never happen? We are reading from the changelog, to
if there is no headers, we would get empty headers, but never `null`? So not
sure if we need this check? `headers.lastHeader` should cover the empty case
correctly anyway?
##########
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}.
+ * <p>
+ * 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.
+ * <p>
+ * 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.
+ * <p>
+ * 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 {
Review Comment:
Why does the changelogger implement this interface? Seems it one layer too
high? -- If we would disable changelogging we could get the wrong signal (not
sure if it would break anything, but it seems structurally incorrect -- the
actual RocksDB store should carry the marker interface IMHO
##########
streams/src/main/java/org/apache/kafka/streams/state/internals/ListValueStoreUpgradeUtils.java:
##########
@@ -0,0 +1,238 @@
+/*
+ * 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.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;
+
+import java.io.ByteArrayOutputStream;
+import java.nio.ByteBuffer;
+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).
+ * <p>
+ * 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}:
+ * <ul>
+ * <li>PLAIN: {@code [leftFlag(1B)][rawValue]} (a {@code
LeftOrRightValue})</li>
+ * <li>HEADERS: {@code
[headersSize(varint)][headersBytes][leftFlag(1B)][rawValue]}
+ * (an {@code AggregationWithHeaders<LeftOrRightValue>})</li>
+ * </ul>
+ * A PLAIN element becomes a HEADERS element with <em>empty</em> 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}.
+ * <p>
+ * The HEADERS element format above is the <em>local, on-disk</em> 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 <em>value</em> can stay in the format an old
PLAIN store understands.
+ * <p>
+ * 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.
+ * <p>
+ * Deliberately namespaced to avoid colliding with user headers that ride
along on the record.
+ * A record <em>without</em> 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<byte[]>> 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<byte[]> plainElements =
LIST_SERDE.deserializer().deserialize(null, plainListBlob);
+ final List<byte[]> 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);
+ }
+
+ /**
+ * 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[])}.
+ * <p>
+ * 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<byte[]> headersElements =
LIST_SERDE.deserializer().deserialize(null, headersListBlob);
+ final List<byte[]> 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
Review Comment:
If `ListValueStore never appends null`, why do we need to handle this case?
Isn't this just dead code?
We could of course add a guard and `throw new SerializationException(...)`,
too.
##########
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}.
+ * <p>
+ * 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.
+ * <p>
+ * 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.
+ * <p>
+ * 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<Bytes,
byte[]> 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());
Review Comment:
Yes, we should not leak, by why not just create an empty header? Why do we
need to copy the content from the context? -- Also not sure why we need the
`remove` below? If we have out own copy, why do we need to remove?
##########
streams/src/main/java/org/apache/kafka/streams/state/internals/ListValueStoreUpgradeUtils.java:
##########
@@ -0,0 +1,238 @@
+/*
+ * 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.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;
+
+import java.io.ByteArrayOutputStream;
+import java.nio.ByteBuffer;
+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).
+ * <p>
+ * 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}:
+ * <ul>
+ * <li>PLAIN: {@code [leftFlag(1B)][rawValue]} (a {@code
LeftOrRightValue})</li>
+ * <li>HEADERS: {@code
[headersSize(varint)][headersBytes][leftFlag(1B)][rawValue]}
+ * (an {@code AggregationWithHeaders<LeftOrRightValue>})</li>
+ * </ul>
+ * A PLAIN element becomes a HEADERS element with <em>empty</em> 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}.
+ * <p>
+ * The HEADERS element format above is the <em>local, on-disk</em> 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 <em>value</em> can stay in the format an old
PLAIN store understands.
+ * <p>
+ * 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.
+ * <p>
+ * Deliberately namespaced to avoid colliding with user headers that ride
along on the record.
+ * A record <em>without</em> 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<byte[]>> 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<byte[]> plainElements =
LIST_SERDE.deserializer().deserialize(null, plainListBlob);
+ final List<byte[]> 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);
+ }
+
+ /**
+ * 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[])}.
+ * <p>
+ * 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<byte[]> headersElements =
LIST_SERDE.deserializer().deserialize(null, headersListBlob);
+ final List<byte[]> 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<byte[]> plainElements =
LIST_SERDE.deserializer().deserialize(null, plainListBlob);
+ final List<byte[]> headersElements = new
ArrayList<>(plainElements.size());
+ final ByteBuffer prefixes = ByteBuffer.wrap(elementHeaders);
+
+ for (final byte[] plainElement : plainElements) {
+ final byte[] prefix = readNextHeadersPrefix(prefixes);
+ if (plainElement == null) {
Review Comment:
This is the counter-part question: if `ListStore` never puts `null` values
into the list, we should never get a `plainElement == null` back here?
##########
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 {
Review Comment:
Should this interface implement `HeadersBytesStoreSupplier` ?
##########
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.
+ * <p>
+ * Used solely by {@code StateManagerUtil.converterForStore} to select the
list-aware restore
+ * {@link RecordConverters#rawListValueToHeadersListValue() converter}.
+ * <p>
+ * 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 {
Review Comment:
Does it need to extend `StateStore`? `HeadersBytesStore` doesn't extend
anything either.
##########
streams/src/main/java/org/apache/kafka/streams/state/internals/ListValueStoreUpgradeUtils.java:
##########
@@ -0,0 +1,238 @@
+/*
+ * 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.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;
+
+import java.io.ByteArrayOutputStream;
+import java.nio.ByteBuffer;
+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).
+ * <p>
+ * 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}:
+ * <ul>
+ * <li>PLAIN: {@code [leftFlag(1B)][rawValue]} (a {@code
LeftOrRightValue})</li>
+ * <li>HEADERS: {@code
[headersSize(varint)][headersBytes][leftFlag(1B)][rawValue]}
+ * (an {@code AggregationWithHeaders<LeftOrRightValue>})</li>
+ * </ul>
+ * A PLAIN element becomes a HEADERS element with <em>empty</em> 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}.
+ * <p>
+ * The HEADERS element format above is the <em>local, on-disk</em> 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 <em>value</em> can stay in the format an old
PLAIN store understands.
+ * <p>
+ * 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.
+ * <p>
+ * Deliberately namespaced to avoid colliding with user headers that ride
along on the record.
+ * A record <em>without</em> 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__";
Review Comment:
Should we use such a long header key? It add quite some overhead per record.
##########
streams/src/main/java/org/apache/kafka/streams/state/internals/ListValueStoreUpgradeUtils.java:
##########
@@ -0,0 +1,238 @@
+/*
+ * 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.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;
+
+import java.io.ByteArrayOutputStream;
+import java.nio.ByteBuffer;
+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).
+ * <p>
+ * 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}:
+ * <ul>
+ * <li>PLAIN: {@code [leftFlag(1B)][rawValue]} (a {@code
LeftOrRightValue})</li>
+ * <li>HEADERS: {@code
[headersSize(varint)][headersBytes][leftFlag(1B)][rawValue]}
+ * (an {@code AggregationWithHeaders<LeftOrRightValue>})</li>
+ * </ul>
+ * A PLAIN element becomes a HEADERS element with <em>empty</em> 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}.
+ * <p>
+ * The HEADERS element format above is the <em>local, on-disk</em> 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 <em>value</em> can stay in the format an old
PLAIN store understands.
+ * <p>
+ * 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.
+ * <p>
+ * Deliberately namespaced to avoid colliding with user headers that ride
along on the record.
+ * A record <em>without</em> 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<byte[]>> 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<byte[]> plainElements =
LIST_SERDE.deserializer().deserialize(null, plainListBlob);
+ final List<byte[]> 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);
+ }
+
+ /**
+ * 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[])}.
+ * <p>
+ * 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<byte[]> headersElements =
LIST_SERDE.deserializer().deserialize(null, headersListBlob);
+ final List<byte[]> 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()
Review Comment:
Seems we could put a small optimization: if all headers are actually empty,
we could just return `null` here throwing away the buffer?
##########
streams/src/main/java/org/apache/kafka/streams/state/internals/ListValueStoreUpgradeUtils.java:
##########
@@ -0,0 +1,238 @@
+/*
+ * 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.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;
+
+import java.io.ByteArrayOutputStream;
+import java.nio.ByteBuffer;
+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).
+ * <p>
+ * 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}:
+ * <ul>
+ * <li>PLAIN: {@code [leftFlag(1B)][rawValue]} (a {@code
LeftOrRightValue})</li>
+ * <li>HEADERS: {@code
[headersSize(varint)][headersBytes][leftFlag(1B)][rawValue]}
+ * (an {@code AggregationWithHeaders<LeftOrRightValue>})</li>
+ * </ul>
+ * A PLAIN element becomes a HEADERS element with <em>empty</em> 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}.
+ * <p>
+ * The HEADERS element format above is the <em>local, on-disk</em> 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 <em>value</em> can stay in the format an old
PLAIN store understands.
+ * <p>
+ * 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.
+ * <p>
+ * Deliberately namespaced to avoid colliding with user headers that ride
along on the record.
+ * A record <em>without</em> 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<byte[]>> 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<byte[]> plainElements =
LIST_SERDE.deserializer().deserialize(null, plainListBlob);
+ final List<byte[]> 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);
+ }
+
+ /**
+ * 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[])}.
+ * <p>
+ * 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<byte[]> headersElements =
LIST_SERDE.deserializer().deserialize(null, headersListBlob);
+ final List<byte[]> 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;
+ }
Review Comment:
```suggestion
} else {
```
##########
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}.
+ * <p>
+ * 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.
+ * <p>
+ * 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.
+ * <p>
+ * 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<Bytes,
byte[]> 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));
Review Comment:
Why do we call `changelogHeaders(null)`? Should we not just pass in
`internalContext.recordContext().headers()` ? -- At least this is what we do on
the other header store.
Or do we actually need a copy as your comment in `changelogHeaders(...)`
says, and all other stores have a bug?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
