aliehsaeedii commented on code in PR #21996: URL: https://github.com/apache/kafka/pull/21996#discussion_r3557764503
########## streams/src/main/java/org/apache/kafka/streams/state/internals/LazyHeaders.java: ########## @@ -0,0 +1,149 @@ +/* + * 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.Header; +import org.apache.kafka.common.header.Headers; +import org.apache.kafka.common.header.internals.RecordHeader; +import org.apache.kafka.common.header.internals.RecordHeaders; + +import java.util.ArrayList; +import java.util.Arrays; +import java.util.Iterator; +import java.util.List; +import java.util.Objects; + +/** + * A lazy implementation of {@link Headers} that defers deserialization of header bytes + * until first read access. This avoids unnecessary parsing when the downstream + * deserializer does not inspect headers. + * + * <p>Headers added via {@link #add(Header)} or {@link #add(String, byte[])} before + * materialization are accumulated in a side list and merged on first read access. + * + * <p>Instances are confined to a single {@code StreamThread} and are not shared + * across threads, so no synchronization is needed. + */ +class LazyHeaders implements Headers { + + private final byte[] rawHeaders; + private RecordHeaders materialized; + private List<Header> pendingAdds; + + /** + * Creates a new LazyHeaders wrapping the given raw header bytes. + * + * @param rawHeaders the serialized header bytes (without the varint size prefix), + * as expected by {@link HeadersDeserializer#deserialize(byte[])}. + * May be null or empty for empty headers. + */ + LazyHeaders(final byte[] rawHeaders) { + this.rawHeaders = rawHeaders; + } + + private RecordHeaders materialize() { + if (materialized == null) { + final Headers deserialized = HeadersDeserializer.deserialize(rawHeaders); + materialized = (deserialized instanceof RecordHeaders) + ? (RecordHeaders) deserialized + : new RecordHeaders(deserialized); + if (pendingAdds != null) { + for (final Header h : pendingAdds) { + materialized.add(h); + } + pendingAdds = null; + } + } + return materialized; + } + + /** + * Returns true if the headers have been deserialized. + * Visible for testing. + */ + boolean isDeserialized() { + return materialized != null; + } + + @Override + public Headers add(final Header header) throws IllegalStateException { + Objects.requireNonNull(header, "header cannot be null"); + if (materialized != null) { + materialized.add(header); + } else { + if (pendingAdds == null) { + pendingAdds = new ArrayList<>(); + } + pendingAdds.add(header); + } + return this; + } + + @Override + public Headers add(final String key, final byte[] value) throws IllegalStateException { + return add(new RecordHeader(key, value)); + } + + @Override + public Headers remove(final String key) throws IllegalStateException { + materialize().remove(key); + return this; + } + + @Override + public Header lastHeader(final String key) { + return materialize().lastHeader(key); + } + + @Override + public Iterable<Header> headers(final String key) { + return materialize().headers(key); + } + + @Override + public Header[] toArray() { + return materialize().toArray(); + } + + @Override + public Iterator<Header> iterator() { + return materialize().iterator(); + } + + @Override + public boolean equals(final Object o) { + if (this == o) return true; + if (!(o instanceof Headers)) return false; Review Comment: `LazyHeaders.equals` compares content against any `Headers`, but `RecordHeaders.equals` uses a `getClass()` check — so `lazy.equals(recordHeaders)` can be true while `recordHeaders.equals(lazy)` is always false. That breaks the equals symmetry contract and misbehaves in sets/maps and Mockito arg matching (which is why the changelog test had to switch to `argThat`). ########## streams/src/main/java/org/apache/kafka/streams/state/internals/LazyHeaders.java: ########## @@ -0,0 +1,149 @@ +/* + * 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.Header; +import org.apache.kafka.common.header.Headers; +import org.apache.kafka.common.header.internals.RecordHeader; +import org.apache.kafka.common.header.internals.RecordHeaders; + +import java.util.ArrayList; +import java.util.Arrays; +import java.util.Iterator; +import java.util.List; +import java.util.Objects; + +/** + * A lazy implementation of {@link Headers} that defers deserialization of header bytes + * until first read access. This avoids unnecessary parsing when the downstream + * deserializer does not inspect headers. + * + * <p>Headers added via {@link #add(Header)} or {@link #add(String, byte[])} before + * materialization are accumulated in a side list and merged on first read access. + * + * <p>Instances are confined to a single {@code StreamThread} and are not shared + * across threads, so no synchronization is needed. + */ +class LazyHeaders implements Headers { + + private final byte[] rawHeaders; + private RecordHeaders materialized; + private List<Header> pendingAdds; + + /** + * Creates a new LazyHeaders wrapping the given raw header bytes. + * + * @param rawHeaders the serialized header bytes (without the varint size prefix), + * as expected by {@link HeadersDeserializer#deserialize(byte[])}. + * May be null or empty for empty headers. + */ + LazyHeaders(final byte[] rawHeaders) { + this.rawHeaders = rawHeaders; + } + + private RecordHeaders materialize() { + if (materialized == null) { + final Headers deserialized = HeadersDeserializer.deserialize(rawHeaders); + materialized = (deserialized instanceof RecordHeaders) Review Comment: `HeadersDeserializer.deserialize` always returns a `RecordHeaders`, so this `instanceof` check is always true and the `new RecordHeaders(deserialized)` branch is unreachable — a plain cast works. ########## streams/src/main/java/org/apache/kafka/streams/state/internals/ValueTimestampHeadersDeserializer.java: ########## @@ -73,7 +74,7 @@ public ValueTimestampHeaders<V> deserialize(final String topic, final byte[] val final int headersSize = ByteUtils.readVarint(buffer); final byte[] rawHeaders = readBytes(buffer, headersSize); - final Headers headers = HeadersDeserializer.deserialize(rawHeaders); + final Headers headers = (headersSize == 0) ? new RecordHeaders() : new LazyHeaders(rawHeaders); Review Comment: This repeats the `headersSize == 0 ? new RecordHeaders() : new LazyHeaders(...)` logic that `Utils.readHeaders(buffer)` already does — the sibling `AggregationWithHeadersDeserializer` just calls `Utils.readHeaders(buffer)`. Reuse it here so the two copies can't drift. ########## streams/src/main/java/org/apache/kafka/streams/state/internals/Utils.java: ########## @@ -173,7 +173,7 @@ public static byte[] readBytes(final ByteBuffer buffer, final int length) { public static Headers readHeaders(final ByteBuffer buffer) { final int headersSize = ByteUtils.readVarint(buffer); final byte[] rawHeaders = readBytes(buffer, headersSize); - return HeadersDeserializer.deserialize(rawHeaders); + return (headersSize == 0) ? new RecordHeaders() : new LazyHeaders(rawHeaders); Review Comment: **Main structural concern — this leaks the optimization into the changelog *write* path, which is out of scope for KAFKA-20155 and collides with KAFKA-20179 (#21676).** `Utils.readHeaders(buffer)` is shared by two very different callers: - `AggregationWithHeadersDeserializer.deserialize` — the session **read** path (correct; laziness genuinely avoids the parse when the value deserializer ignores headers). - `Utils.headers(...)` — the `ChangeLogging*WithHeaders` **write/changelog** path. On the write path, laziness buys ~nothing: `RecordCollectorImpl.send` builds a `ProducerRecord`, whose constructor does `new RecordHeaders(headers)` and iterates the `LazyHeaders` immediately, forcing materialization before the record even reaches the producer. So here you only add wrapper overhead, and you're editing the same seam that #21676 rewrites. Suggestion: keep 20155 read-only. Introduce `LazyHeaders` at the read `deserialize(...)` entry points (a dedicated `lazyReadHeaders` helper is fine, and addresses the DRY point on `ValueTimestampHeadersDeserializer` too), but leave `Utils.headers()` / the write path on eager `HeadersDeserializer.deserialize`. That keeps 20155 and 20179 cleanly separable. ########## streams/src/main/java/org/apache/kafka/streams/state/AggregationWithHeaders.java: ########## @@ -96,12 +97,18 @@ public boolean equals(final Object o) { } final AggregationWithHeaders<?> that = (AggregationWithHeaders<?>) o; return Objects.equals(aggregation, that.aggregation) - && Objects.equals(this.headers, that.headers); + && headersEqual(this.headers, that.headers); + } + + private static boolean headersEqual(final Headers a, final Headers b) { + if (a == b) return true; + if (a == null || b == null) return false; Review Comment: The constructor rejects null headers via `requireNonNull`, so `a == null || b == null` can never be true here — dead check. (The `make`/`makeAllowNullable` javadoc saying null is allowed is stale too.) ########## streams/src/main/java/org/apache/kafka/streams/state/ValueTimestampHeaders.java: ########## @@ -108,12 +109,18 @@ public boolean equals(final Object o) { final ValueTimestampHeaders<?> that = (ValueTimestampHeaders<?>) o; return timestamp == that.timestamp && Objects.equals(value, that.value) - && Objects.equals(this.headers, that.headers); + && headersEqual(this.headers, that.headers); + } + + private static boolean headersEqual(final Headers a, final Headers b) { + if (a == b) return true; + if (a == null || b == null) return false; + return Arrays.equals(a.toArray(), b.toArray()); } @Override public int hashCode() { - return Objects.hash(value, timestamp, headers); + return Objects.hash(value, timestamp, Arrays.hashCode(headers.toArray())); Review Comment: Two notes on this `equals`/`hashCode` change (applies symmetrically to `AggregationWithHeaders`): 1. **It forces materialization.** `toArray()` parses the `LazyHeaders`, so comparing or hashing a `ValueTimestampHeaders`/`AggregationWithHeaders` defeats the laziness this PR adds. Mostly a test concern, but worth confirming no internal path hashes/dedups these value objects. 2. **These are public `o.a.k.streams.state` classes**, so this is a public behavioral change: header equality goes from `RecordHeaders` list-equals to `toArray()` content-equals, and `hashCode` values change. It's functionally equivalent (both order-sensitive) and arguably more correct, but please call it out in the PR description. Root cause is the `LazyHeaders.equals` asymmetry flagged separately — if that's addressed, these two public types may not need to change at all. -- 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]
