aliehsaeedii commented on code in PR #22961: URL: https://github.com/apache/kafka/pull/22961#discussion_r3667444642
########## 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: Correct, same as above. Added the matching guard. The `convertPlainListBlobToHeadersListBlob` is the same — Should we make it strict as well? -- 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]
