aliehsaeedii commented on code in PR #22961: URL: https://github.com/apache/kafka/pull/22961#discussion_r3667416362
########## 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: Dead code, agreed. ListValueStore is the only writer: put/putIfAbsent turn a null value into a whole-list delete and never reach the list, and putAll throws. Replaced with a SerializationException guard so a future regression fails loudly instead of NPE-ing in ByteBuffer.wrap -- 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]
