mjsax commented on code in PR #22165:
URL: https://github.com/apache/kafka/pull/22165#discussion_r3661685733
##########
streams/src/test/java/org/apache/kafka/streams/state/internals/TimeOrderedKeyValueBufferTest.java:
##########
@@ -306,10 +306,74 @@ public void shouldReturnPriorValueForBufferedKey(final
String testName, final Fu
context.setRecordContext(recordContext);
buffer.put(1L, new Record<>("A", new Change<>("new-value",
"old-value"), 0L), recordContext);
buffer.put(1L, new Record<>("B", new Change<>("new-value", null), 0L),
recordContext);
- assertThat(buffer.priorValueForBuffered("A"),
is(Maybe.defined(ValueTimestampHeaders.make("old-value", -1, new
RecordHeaders()))));
+ assertThat(buffer.priorValueForBuffered("A"),
is(Maybe.defined(ValueTimestampHeaders.make("old-value", 0L, new
RecordHeaders()))));
assertThat(buffer.priorValueForBuffered("B"), is(Maybe.defined(null)));
}
+ @ParameterizedTest
+ @MethodSource("parameters")
+ public void shouldPropagateHeadersThroughEviction(final String testName,
final Function<String, B> bufferSupplier) {
+ setup(testName, bufferSupplier);
+ final TimeOrderedKeyValueBuffer<String, String, Change<String>> buffer
= bufferSupplier.apply(testName);
+ final MockInternalProcessorContext<?, ?> context = makeContext();
+ buffer.init(context, buffer);
+
+ final RecordHeaders headers = new RecordHeaders(new Header[]{new
RecordHeader("h1", "v1".getBytes(UTF_8))});
Review Comment:
It's always better to specify it, to stay platform independent.
##########
streams/src/test/java/org/apache/kafka/streams/kstream/internals/SuppressHeadersScenarioTest.java:
##########
@@ -0,0 +1,616 @@
+/*
+ * 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.kstream.internals;
+
+import org.apache.kafka.common.header.Header;
+import org.apache.kafka.common.header.Headers;
+import org.apache.kafka.common.header.internals.RecordHeaders;
+import org.apache.kafka.common.serialization.Deserializer;
+import org.apache.kafka.common.serialization.Serde;
+import org.apache.kafka.common.serialization.Serdes;
+import org.apache.kafka.common.serialization.Serializer;
+import org.apache.kafka.common.serialization.StringDeserializer;
+import org.apache.kafka.common.serialization.StringSerializer;
+import org.apache.kafka.common.utils.Bytes;
+import org.apache.kafka.streams.KeyValue;
+import org.apache.kafka.streams.StreamsBuilder;
+import org.apache.kafka.streams.StreamsConfig;
+import org.apache.kafka.streams.TestInputTopic;
+import org.apache.kafka.streams.Topology;
+import org.apache.kafka.streams.TopologyTestDriver;
+import org.apache.kafka.streams.TopologyTestDriverBuilder;
+import org.apache.kafka.streams.kstream.Consumed;
+import org.apache.kafka.streams.kstream.Grouped;
+import org.apache.kafka.streams.kstream.KTable;
+import org.apache.kafka.streams.kstream.Materialized;
+import org.apache.kafka.streams.kstream.Produced;
+import org.apache.kafka.streams.kstream.TimeWindows;
+import org.apache.kafka.streams.kstream.Windowed;
+import org.apache.kafka.streams.state.KeyValueStore;
+import org.apache.kafka.streams.state.WindowStore;
+import org.apache.kafka.streams.test.TestRecord;
+import org.apache.kafka.test.TestUtils;
+
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+import java.nio.charset.StandardCharsets;
+import java.time.Duration;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Locale;
+import java.util.Objects;
+import java.util.Properties;
+
+import static java.time.Duration.ofMillis;
+import static
org.apache.kafka.streams.kstream.Suppressed.BufferConfig.unbounded;
+import static org.apache.kafka.streams.kstream.Suppressed.untilTimeLimit;
+import static org.apache.kafka.streams.kstream.Suppressed.untilWindowCloses;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+/**
+ * Scenario tests for record-header propagation through {@code suppress()}.
+ *
+ * <p>The suppress buffer holds a row per key containing an {@code old} and a
{@code new} value,
+ * plus a single record context. Each of those parts originates from a
+ * <em>different</em> input record, so each must be serialized and
deserialized with the headers of
+ * the record it came from. Getting this wrong is invisible with
header-agnostic serdes such as
+ * {@link Serdes#String()}, because they discard the headers they are handed.
These tests therefore
+ * use serdes that observe every invocation, and assert on what the serdes
actually saw rather than
+ * only on what reached the output topic.
+ *
+ * <p>Each input record carries the headers it expects to see preserved, as
{@code origin=<value>},
+ * so every value should always meet its own marker; see {@link
#observingSerde}.
+ *
+ * <h3>Invariants asserted</h3>
+ * <ul>
+ * <li><b>INV-1</b> — every value (de)serialized by the suppress buffer must
be handed the headers
+ * belonging to that value, i.e. {@code origin=<value>}.</li>
+ * <li><b>INV-2</b> — the record emitted on eviction must carry the headers
belonging to the value
+ * being emitted.</li>
+ * <li><b>INV-3</b> — reported, not enforced: which headers the suppress
buffer hands to the
+ * <em>key</em> serde. The key is shared by all parts of a row, so there
is no single "correct"
+ * answer to pin down yet; the observations are printed so the behaviour
is visible.</li>
+ * </ul>
+ *
+ * <p>All checks are collected and reported together: a scenario runs to
completion, prints a full
+ * trace plus a check summary to stdout, and only then makes a single
assertion. That keeps the whole
+ * picture visible instead of stopping at the first deviation.
+ *
+ * <h3>Scenario coverage</h3>
+ *
+ * The buffer never sees {@code Windowed}: it stores serialized key bytes, and
windowing reaches
+ * {@code suppress()} only through the time definition and through {@code
safeToDropTombstones},
+ * neither of which affects header handling. Windowed and non-windowed
scenarios therefore drive the
+ * same code, and the matrix below is deliberately not filled in exhaustively.
+ *
+ * <p>In particular there is <b>no</b> windowed /
eviction-triggered-by-a-different-key scenario. That
+ * is omitted on purpose rather than missing: it would drive the same buffer
path as
+ * {@link #nonWindowedEvictionTriggeredByDifferentKey()} with nothing added.
The windowed scenarios
+ * that are here earn their place either by recording row-identity semantics
that are easy to get
+ * wrong, or by being reachable only when windowed.
+ *
+ * <h3>Why the same-row scenarios require the HEADERS store format</h3>
+ *
+ * When the evicting record updates a row that is already buffered, that row
ends up holding an
+ * {@code old} and a {@code new} value originating from two different records,
so INV-1 requires two
+ * different sets of headers to be recoverable from one row. The plain format
has nowhere to put
+ * them: it stores bare values and a single record context, so at most one of
the two can be right.
+ * INV-1 is therefore unsatisfiable in plain format for these scenarios, no
matter how
+ * {@code suppress()} is implemented.
+ *
+ * <p>{@link #nonWindowedEvictionTriggeredBySameKey()},
+ * {@link #windowedEvictionTriggeredBySameKeySameWindow()},
+ * {@link #nonWindowedTombstoneMustBeForwarded()} and
+ * {@link #windowedFinalResultsTombstoneIsDropped()} consequently set
+ * {@code dsl.store.format=headers}, where each value part carries its own
headers. Running them in
+ * plain format would assert something impossible and read as a permanent bug.
The two tombstone
+ * scenarios belong in that group as well: a delete overwrites the row it
lands on, so the row still
+ * holds an {@code old} value from the earlier record alongside the {@code
new} null.
+ *
+ * <p>Scenarios where the evicted row is <em>not</em> the one being updated
have only one origin per
+ * row, so they are satisfiable in plain format and deliberately stay there.
+ */
+public class SuppressHeadersScenarioTest {
Review Comment:
The other PR was merged, so this class should drop after a rebase -- we
still need to enable the tests of course.
##########
streams/src/main/java/org/apache/kafka/streams/state/internals/InMemoryTimeOrderedKeyValueChangeBuffer.java:
##########
@@ -203,6 +222,10 @@ public void init(final StateStoreContext
stateStoreContext, final StateStore roo
taskId = context.taskId().toString();
streamsMetrics = context.metrics();
+ final Object dslStoreFormat =
stateStoreContext.appConfigs().get(StreamsConfig.DSL_STORE_FORMAT_CONFIG);
Review Comment:
Is there a good place in the docs, to call this out? -- Should we file a
Jira ticket about "allow enabling header-support on suppress() similar to
`Materialized` for other operators"?
##########
streams/src/main/java/org/apache/kafka/streams/state/internals/InMemoryTimeOrderedKeyValueChangeBuffer.java:
##########
@@ -73,6 +75,12 @@ public final class
InMemoryTimeOrderedKeyValueChangeBuffer<K, V, T> implements T
private static final byte[] V_3_CHANGELOG_HEADER_VALUE = {(byte) 3};
static final RecordHeaders CHANGELOG_HEADERS =
new RecordHeaders(new Header[] {new RecordHeader("v",
V_3_CHANGELOG_HEADER_VALUE)});
+ // Each buffered value part's headers/timestamp travel in the changelog
record's own Kafka headers -- one per part,
+ // holding that part's [headersSize][headers][timestamp] prefix -- so the
value bytes stay V3 and remain restorable
+ // by older versions, which simply ignore these headers.
+ static final String PRIOR_VALUE_HEADERS_KEY = "vh.prior";
+ static final String OLD_VALUE_HEADERS_KEY = "vh.old";
+ static final String NEW_VALUE_HEADERS_KEY = "vh.new";
Review Comment:
We might not need this last one after all? The `context` that we already
encode should contain the `newValue` header and timestamp already, so no need
to put into the headers a second time?
##########
streams/src/main/java/org/apache/kafka/streams/state/internals/InMemoryTimeOrderedKeyValueChangeBuffer.java:
##########
@@ -465,28 +512,136 @@ private byte[] internalPriorValueForBuffered(final Bytes
key) {
}
}
+ // The legacy V0/V1 restore paths feed the currently-buffered prior value
back in as plain value
+ // bytes, so unwrap it when the in-memory encoding is the headers-aware
format.
+ private byte[] plainPriorValueForBuffered(final Bytes key) {
+ final byte[] priorValue = internalPriorValueForBuffered(key);
+ return storeHeaders ? Utils.rawPlainValue(priorValue) : priorValue;
+ }
+
+ // Normalizes a restored buffer value so its value parts match the
encoding currently used
+ // in-memory (ValueTimestampHeaders blobs when header stores are enabled,
plain values otherwise),
+ // regardless of the changelog version it was read from.
+ private BufferValue normalizeEncoding(final BufferValue value, final
boolean restoredWithHeaders) {
+ if (storeHeaders == restoredWithHeaders) {
+ return value;
+ }
+ if (storeHeaders) {
+ // Plain -> ValueTimestampHeaders. The original headers/timestamp
of legacy records are
+ // unknown, so we use empty headers and the record-context
timestamp.
+ final long timestamp = value.context().timestamp();
+ return new BufferValue(
+ Utils.rawValueTimestampHeaders(value.priorValue(), timestamp),
+ Utils.rawValueTimestampHeaders(value.oldValue(), timestamp),
+ Utils.rawValueTimestampHeaders(value.newValue(), timestamp),
+ value.context()
+ );
+ }
+ // ValueTimestampHeaders -> plain would drop the per-value
headers/timestamps carried by a V4
+ // changelog. This only arises when a store that previously used a
headers store format is
+ // restored without one, i.e. a store-format downgrade. Downgrades are
not supported, so we
+ // fail fast here rather than silently discarding the stored header
data.
+ throw new StreamsException(
+ "Cannot restore suppress buffer '" + storeName + "': its changelog
contains header-aware "
+ + "(V4) records, but the store is configured without a headers
store format. "
+ + "Downgrading the store format is not supported."
+ );
+ }
+
+ private ValueTimestampHeadersSerializer<V>
valueTimestampHeadersSerializer() {
+ if (valueTimestampHeadersSerializer == null) {
+ valueTimestampHeadersSerializer = new
ValueTimestampHeadersSerializer<>(valueSerde.innerSerde().serializer());
+ }
+ return valueTimestampHeadersSerializer;
+ }
+
+ private ValueTimestampHeadersDeserializer<V>
valueTimestampHeadersDeserializer() {
+ if (valueTimestampHeadersDeserializer == null) {
+ valueTimestampHeadersDeserializer = new
ValueTimestampHeadersDeserializer<>(valueSerde.innerSerde().deserializer());
+ }
+ return valueTimestampHeadersDeserializer;
+ }
+
+ // Serialize a single value part. When storeHeaders is set, the value is
wrapped as a
+ // ValueTimestampHeaders blob carrying the given timestamp and headers;
otherwise the plain
+ // value bytes are stored (the pre-existing V3 behavior). Returns null for
a null value.
+ private byte[] serializeValuePart(final V value, final long timestamp,
final Headers headers) {
+ if (value == null) {
+ return null;
+ }
+ if (storeHeaders) {
+ return valueTimestampHeadersSerializer().serialize(changelogTopic,
ValueTimestampHeaders.make(value, timestamp, headers));
+ }
+ return valueSerde.innerSerde().serializer().serialize(changelogTopic,
headers, value);
+ }
+
+ // Deserialize a single stored value part into a ValueTimestampHeaders.
Only valid when
+ // storeHeaders is set (i.e. the part bytes are a ValueTimestampHeaders
blob).
+ private ValueTimestampHeaders<V> deserializeValuePart(final byte[] bytes) {
+ return bytes == null ? null :
valueTimestampHeadersDeserializer().deserialize(changelogTopic, bytes);
+ }
+
+ // Deserialize a single stored value part into the plain value, handling
both the headers-aware
+ // (ValueTimestampHeaders) and plain encodings.
+ private V deserializeValue(final byte[] bytes, final Headers
fallbackHeaders) {
+ if (bytes == null) {
+ return null;
+ }
+ if (storeHeaders) {
+ return
ValueTimestampHeaders.getValueOrNull(deserializeValuePart(bytes));
+ }
+ return
valueSerde.innerSerde().deserializer().deserialize(changelogTopic,
fallbackHeaders, bytes);
+ }
+
@Override
public boolean put(final long time,
final Record<K, Change<V>> record,
final ProcessorRecordContext recordContext) {
requireNonNull(record.value(), "value cannot be null");
requireNonNull(recordContext, "recordContext cannot be null");
- final Bytes serializedKey =
Bytes.wrap(keySerde.serializer().serialize(changelogTopic,
recordContext.headers(), record.key()));
- final Change<byte[]> serialChange =
valueSerde.serializeParts(changelogTopic, recordContext.headers(),
record.value());
-
+ // The record's own headers (not the processing context's) describe
the new value and must be
+ // the ones forwarded for this key on eviction.
+ final RecordHeaders headers = new RecordHeaders(record.headers());
+ final long timestamp = record.timestamp();
+ final Bytes serializedKey =
Bytes.wrap(keySerde.serializer().serialize(changelogTopic, headers,
record.key()));
final BufferValue buffered = getBuffered(serializedKey);
- final byte[] serializedPriorValue;
- if (buffered == null) {
- serializedPriorValue = serialChange.oldValue;
- } else {
- serializedPriorValue = buffered.priorValue();
+
+ // The context stored with the entry carries the currently-processed
record's headers; these
+ // are what get forwarded downstream when the (new) value is emitted.
Everything else is taken
+ // from the processing context, unchanged from before.
+ final ProcessorRecordContext effectiveContext = new
ProcessorRecordContext(
+ recordContext.timestamp(),
+ recordContext.offset(),
+ recordContext.partition(),
+ recordContext.topic(),
+ headers
+ );
+
+ // The old value's original headers/timestamp are not carried by the
incoming record. On an
+ // in-place update we recover them from the entry's previous new value
(whose value is exactly
+ // this update's old value); on the first insert for a key they are
genuinely unknown.
+ Headers oldHeaders = new RecordHeaders();
+ long oldTimestamp = RecordQueue.UNKNOWN;
+ if (storeHeaders && buffered != null) {
+ final ValueTimestampHeaders<V> previousNewValue =
deserializeValuePart(buffered.newValue());
+ if (previousNewValue != null) {
+ oldHeaders = previousNewValue.headers();
+ oldTimestamp = previousNewValue.timestamp();
+ }
}
+ final Change<V> change = record.value();
+ final byte[] newValue = serializeValuePart(change.newValue, timestamp,
headers);
+ // In plain mode the old value is still serialized with the current
record's headers, exactly
+ // as before, so the stored bytes are unchanged for non-header stores.
+ final byte[] oldValue = serializeValuePart(change.oldValue,
oldTimestamp, storeHeaders ? oldHeaders : headers);
Review Comment:
I think the PR is correct. We should not "pollute" the newValue headers with
stuff from oldValue, to having oldValue isolated in headers mode, is actually a
fix IMHO.
##########
streams/src/main/java/org/apache/kafka/streams/state/internals/InMemoryTimeOrderedKeyValueChangeBuffer.java:
##########
@@ -399,14 +448,43 @@ public void evictWhile(final Supplier<Boolean> predicate,
next.getKey().time() + "]"
);
}
- final K key =
keySerde.deserializer().deserialize(changelogTopic, context.headers(),
next.getKey().key().get());
final BufferValue bufferValue = next.getValue();
- final Change<V> value = valueSerde.deserializeParts(
- changelogTopic,
- context.headers(),
- new Change<>(bufferValue.newValue(),
bufferValue.oldValue())
- );
- callback.accept(new Eviction<K, Change<V>>(key, value,
bufferValue.context()));
+ final ProcessorRecordContext bufferedContext =
bufferValue.context();
+
+ final K key;
+ final Change<V> value;
+ final ProcessorRecordContext evictionContext;
+ if (storeHeaders) {
+ // Each value part was stored as a ValueTimestampHeaders
blob carrying the headers of
+ // the record it originated from, so take them from the
value rather than from the
+ // buffered context: the old value came from an earlier
record than the new one, and
+ // deserializing it with the new record's headers would
hand the serde the wrong ones.
+ // (ValueTimestampHeadersDeserializer feeds each part's
own headers to the inner
+ // value deserializer.)
+ final ValueTimestampHeaders<V> newPart =
deserializeValuePart(bufferValue.newValue());
Review Comment:
I find the name `deserializeValuePart` confusing. We are getting back a vth
object, and we only pass in the corresponding rawVTH byte[] array. Not sure
what "part" refers to?
##########
streams/src/main/java/org/apache/kafka/streams/state/internals/InMemoryTimeOrderedKeyValueChangeBuffer.java:
##########
@@ -465,28 +569,135 @@ private byte[] internalPriorValueForBuffered(final Bytes
key) {
}
}
+ // The legacy V0/V1 restore paths feed the currently-buffered prior value
back in as plain value
+ // bytes, so unwrap it when the in-memory encoding is the headers-aware
format.
+ private byte[] plainPriorValueForBuffered(final Bytes key) {
+ final byte[] priorValue = internalPriorValueForBuffered(key);
+ return storeHeaders ? Utils.rawPlainValue(priorValue) : priorValue;
+ }
+
+ // The changelog value format is plain for every version, so a restored
buffer value always has
+ // plain value parts. When header stores are enabled the in-memory
encoding is
+ // ValueTimestampHeaders, so re-attach each part's headers and timestamp
from the changelog
+ // record's own Kafka headers. Records written before this feature -- or
by a run configured
+ // without header stores -- carry no such headers, in which case the
originals are genuinely
+ // unknown and we fall back to empty headers with the record-context
timestamp.
+ private BufferValue toInMemoryEncoding(final BufferValue value, final
Headers recordHeaders) {
+ if (!storeHeaders) {
+ return value;
+ }
+ final long fallbackTimestamp = value.context().timestamp();
+ return new BufferValue(
+ valuePartWithHeaders(value.priorValue(), recordHeaders,
PRIOR_VALUE_HEADERS_KEY, fallbackTimestamp),
+ valuePartWithHeaders(value.oldValue(), recordHeaders,
OLD_VALUE_HEADERS_KEY, fallbackTimestamp),
+ valuePartWithHeaders(value.newValue(), recordHeaders,
NEW_VALUE_HEADERS_KEY, fallbackTimestamp),
+ value.context()
+ );
+ }
+
+ private static byte[] valuePartWithHeaders(final byte[] rawPlainValue,
+ final Headers recordHeaders,
+ final String headerKey,
+ final long fallbackTimestamp) {
+ if (rawPlainValue == null) {
+ return null;
+ }
+ final Header prefix = recordHeaders.lastHeader(headerKey);
Review Comment:
Why `prefix`? The name is confusing. We store ts+headers in the `Header` to
should this not be `tsAndHeader` as variable name?
--
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]
