This is an automated email from the ASF dual-hosted git repository.
mjsax pushed a commit to branch trunk
in repository https://gitbox.apache.org/repos/asf/kafka.git
The following commit(s) were added to refs/heads/trunk by this push:
new b706c58dd91 KAFKA-20413: Make Streams suppress headers-aware (#22165)
b706c58dd91 is described below
commit b706c58dd91f2f331a91d8b94b15b0912d1a8242
Author: Alieh Saeedi <[email protected]>
AuthorDate: Wed Jul 29 08:17:54 2026 +0200
KAFKA-20413: Make Streams suppress headers-aware (#22165)
This PR fixes the suppress-buffer part of KAFKA-20413 / KIP-1285.
`InMemoryTimeOrderedKeyValueChangeBuffer` now uses the buffered record's
own headers in serde calls and on eviction.
Testing: Added `shouldPropagateHeadersThroughEviction`, and
`shouldUseRecordHeadersNotProcessorContextHeadersOnPut`.
Reviewers: Matthias J. Sax <[email protected]>, Uladzislau Blok
<[email protected]>
---
.../InMemoryTimeOrderedKeyValueChangeBuffer.java | 301 +++++++++--
.../kafka/streams/state/internals/Utils.java | 83 +++
.../internals/SuppressHeadersScenarioTest.java | 28 +-
.../internals/TimeOrderedKeyValueBufferTest.java | 597 +++++++++++++++++++++
.../kafka/streams/state/internals/UtilsTest.java | 74 +++
5 files changed, 1039 insertions(+), 44 deletions(-)
diff --git
a/streams/src/main/java/org/apache/kafka/streams/state/internals/InMemoryTimeOrderedKeyValueChangeBuffer.java
b/streams/src/main/java/org/apache/kafka/streams/state/internals/InMemoryTimeOrderedKeyValueChangeBuffer.java
index 04e6911d1f7..449f0240faa 100644
---
a/streams/src/main/java/org/apache/kafka/streams/state/internals/InMemoryTimeOrderedKeyValueChangeBuffer.java
+++
b/streams/src/main/java/org/apache/kafka/streams/state/internals/InMemoryTimeOrderedKeyValueChangeBuffer.java
@@ -19,6 +19,7 @@ package org.apache.kafka.streams.state.internals;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.common.TopicPartition;
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 org.apache.kafka.common.metrics.Sensor;
@@ -26,6 +27,7 @@ import
org.apache.kafka.common.serialization.ByteArraySerializer;
import org.apache.kafka.common.serialization.BytesSerializer;
import org.apache.kafka.common.serialization.Serde;
import org.apache.kafka.common.utils.Bytes;
+import org.apache.kafka.streams.StreamsConfig;
import org.apache.kafka.streams.kstream.internals.Change;
import org.apache.kafka.streams.kstream.internals.FullChangeSerde;
import org.apache.kafka.streams.processor.StateStore;
@@ -73,6 +75,13 @@ 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)});
+ // The prior and old value parts' 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. The
new value part needs no such header: the
+ // record context encoded in the V3 value already describes it, because
put() captures the context of the very
+ // record the new value came from.
+ static final String PRIOR_VALUE_HEADERS_KEY = "vh.prior";
+ static final String OLD_VALUE_HEADERS_KEY = "vh.old";
private static final String METRIC_SCOPE = "in-memory-suppression";
private final Map<Bytes, BufferKey> index = new HashMap<>();
@@ -85,6 +94,14 @@ public final class
InMemoryTimeOrderedKeyValueChangeBuffer<K, V, T> implements T
private Serde<K> keySerde;
private FullChangeSerde<V> valueSerde;
+ // When headers-aware stores are enabled (dsl.store.format=HEADERS) each
buffered value part is
+ // stored as a ValueTimestampHeaders blob so that the old and new value
can carry their own
+ // headers and timestamp independently. Otherwise plain values are stored
(the pre-existing V3
+ // behavior) to avoid inflating memory/changelog size for users who do not
use header stores.
+ private boolean storeHeaders;
+ private ValueTimestampHeadersSerializer<V> valueTimestampHeadersSerializer;
+ private ValueTimestampHeadersDeserializer<V>
valueTimestampHeadersDeserializer;
+
private long memBufferSize = 0L;
private long minTimestamp = Long.MAX_VALUE;
private InternalProcessorContext<?, ?> context;
@@ -203,6 +220,10 @@ public final class
InMemoryTimeOrderedKeyValueChangeBuffer<K, V, T> implements T
taskId = context.taskId().toString();
streamsMetrics = context.metrics();
+ final Object dslStoreFormat =
stateStoreContext.appConfigs().get(StreamsConfig.DSL_STORE_FORMAT_CONFIG);
+ storeHeaders = dslStoreFormat != null
+ &&
StreamsConfig.DSL_STORE_FORMAT_HEADERS.equalsIgnoreCase(dslStoreFormat.toString());
+
bufferSizeSensor = StateStoreMetrics.suppressionBufferSizeSensor(
taskId,
METRIC_SCOPE,
@@ -268,14 +289,17 @@ public final class
InMemoryTimeOrderedKeyValueChangeBuffer<K, V, T> implements T
private void logValue(final Bytes key, final BufferKey bufferKey, final
BufferValue value) {
final int sizeOfBufferTime = Long.BYTES;
- final ByteBuffer buffer = value.serialize(sizeOfBufferTime);
+ // The logged value is always in the V3 format: in headers mode the
in-memory value parts are
+ // ValueTimestampHeaders blobs, so split each one and log only its
plain value bytes, sending
+ // the headers/timestamp prefixes alongside in the record's Kafka
headers.
+ final ByteBuffer buffer = (storeHeaders ? plainEncoded(value) :
value).serialize(sizeOfBufferTime);
buffer.putLong(bufferKey.time());
final byte[] array = buffer.array();
((RecordCollector.Supplier) context).recordCollector().send(
changelogTopic,
key,
array,
- CHANGELOG_HEADERS,
+ storeHeaders ? changelogHeadersWithValuePartHeaders(value) :
CHANGELOG_HEADERS,
partition,
null,
KEY_SERIALIZER,
@@ -284,6 +308,36 @@ public final class
InMemoryTimeOrderedKeyValueChangeBuffer<K, V, T> implements T
null);
}
+ private static BufferValue plainEncoded(final BufferValue value) {
+ return new BufferValue(
+ Utils.rawPlainValue(value.priorValue()),
+ Utils.rawPlainValue(value.oldValue()),
+ Utils.rawPlainValue(value.newValue()),
+ value.context()
+ );
+ }
+
+ private static RecordHeaders changelogHeadersWithValuePartHeaders(final
BufferValue value) {
+ final RecordHeaders headers = new RecordHeaders(new Header[] {new
RecordHeader("v", V_3_CHANGELOG_HEADER_VALUE)});
+ // BufferValue collapses the prior and old values onto one array
whenever they are equal -- the
+ // common case, since on the first buffering of a key the prior value
IS the old value -- and
+ // serialize() then writes those bytes only once. Skip the duplicate
prefix as well; restoring
+ // recovers the prior part from vh.old whenever vh.prior is absent.
+ if (value.priorValue() != value.oldValue()) {
+ addValuePartHeader(headers, PRIOR_VALUE_HEADERS_KEY,
value.priorValue());
+ }
+ addValuePartHeader(headers, OLD_VALUE_HEADERS_KEY, value.oldValue());
+ return headers;
+ }
+
+ private static void addValuePartHeader(final RecordHeaders headers,
+ final String headerKey,
+ final byte[]
rawValueTimestampHeaders) {
+ if (rawValueTimestampHeaders != null) {
+ headers.add(new RecordHeader(headerKey,
Utils.rawHeadersTimestampPrefix(rawValueTimestampHeaders)));
+ }
+ }
+
private void logTombstone(final Bytes key) {
((RecordCollector.Supplier) context).recordCollector().send(
changelogTopic,
@@ -324,6 +378,7 @@ public final class
InMemoryTimeOrderedKeyValueChangeBuffer<K, V, T> implements T
}
} else {
final Header versionHeader = record.headers().lastHeader("v");
+ final DeserializationResult deserializationResult;
if (versionHeader == null) {
// Version 0:
// value:
@@ -331,10 +386,9 @@ public final class
InMemoryTimeOrderedKeyValueChangeBuffer<K, V, T> implements T
// - old value
// - new value
final byte[] previousBufferedValue = index.containsKey(key)
- ? internalPriorValueForBuffered(key)
+ ? plainPriorValueForBuffered(key)
: null;
- final DeserializationResult deserializationResult =
deserializeV0(record, key, previousBufferedValue);
- cleanPut(deserializationResult.time(),
deserializationResult.key(), deserializationResult.bufferValue());
+ deserializationResult = deserializeV0(record, key,
previousBufferedValue);
} else if (Arrays.equals(versionHeader.value(),
V_3_CHANGELOG_HEADER_VALUE)) {
// Version 3:
// value:
@@ -343,9 +397,7 @@ public final class
InMemoryTimeOrderedKeyValueChangeBuffer<K, V, T> implements T
// - old value
// - new value
// - buffer time
- final DeserializationResult deserializationResult =
deserializeV3(record, key);
- cleanPut(deserializationResult.time(),
deserializationResult.key(), deserializationResult.bufferValue());
-
+ deserializationResult = deserializeV3(record, key);
} else if (Arrays.equals(versionHeader.value(),
V_2_CHANGELOG_HEADER_VALUE)) {
// Version 2:
// value:
@@ -356,8 +408,7 @@ public final class
InMemoryTimeOrderedKeyValueChangeBuffer<K, V, T> implements T
// - buffer time
// NOTE: 2.4.0, 2.4.1, and 2.5.0 actually encode Version 3
formatted data,
// but still set the Version 2 flag, so to deserialize, we
have to duck type.
- final DeserializationResult deserializationResult =
duckTypeV2(record, key);
- cleanPut(deserializationResult.time(),
deserializationResult.key(), deserializationResult.bufferValue());
+ deserializationResult = duckTypeV2(record, key);
} else if (Arrays.equals(versionHeader.value(),
V_1_CHANGELOG_HEADER_VALUE)) {
// Version 1:
// value:
@@ -366,13 +417,17 @@ public final class
InMemoryTimeOrderedKeyValueChangeBuffer<K, V, T> implements T
// - old value
// - new value
final byte[] previousBufferedValue = index.containsKey(key)
- ? internalPriorValueForBuffered(key)
+ ? plainPriorValueForBuffered(key)
: null;
- final DeserializationResult deserializationResult =
deserializeV1(record, key, previousBufferedValue);
- cleanPut(deserializationResult.time(),
deserializationResult.key(), deserializationResult.bufferValue());
+ deserializationResult = deserializeV1(record, key,
previousBufferedValue);
} else {
throw new IllegalArgumentException("Restoring apparently
invalid changelog record: " + record);
}
+ cleanPut(
+ deserializationResult.time(),
+ deserializationResult.key(),
+ toInMemoryEncoding(deserializationResult.bufferValue(),
record.headers())
+ );
}
}
updateBufferMetrics();
@@ -399,14 +454,51 @@ public final class
InMemoryTimeOrderedKeyValueChangeBuffer<K, V, T> implements T
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 =
deserializeValueTimestampHeaders(bufferValue.newValue());
+ final ValueTimestampHeaders<V> oldPart =
deserializeValueTimestampHeaders(bufferValue.oldValue());
+ // A tombstone has no new value to take headers from; fall
back to the buffered context.
+ // That fallback is not a second-best guess: the buffered
context and the new value's
+ // embedded headers always agree. put() hands the very
same record.headers() instance to
+ // both, and ValueTimestampHeadersSerializer snapshots the
headers only AFTER running the
+ // inner value serializer, so even a serde that writes
into the headers it is given (a
+ // Schema Registry serde recording the schema id, say) is
reflected in both. Restoring
+ // preserves the equality, because toInMemoryEncoding
rebuilds the new part from that
+ // same context. Only the old and prior parts can differ
from the context, which is why
+ // they are the ones read from the value here.
+ final Headers newHeaders = newPart == null ?
bufferedContext.headers() : newPart.headers();
+ key = keySerde.deserializer().deserialize(changelogTopic,
newHeaders, next.getKey().key().get());
+ value = new Change<>(
+ ValueTimestampHeaders.getValueOrNull(newPart),
+ ValueTimestampHeaders.getValueOrNull(oldPart)
+ );
+ // The emitted record carries the new value's headers, so
hand them to the callback
+ // via the eviction's context (which is what the suppress
processor forwards).
+ evictionContext = withHeaders(bufferedContext, newHeaders);
+ } else {
+ // Plain values store no headers of their own; the
buffered context is the only
+ // carrier, exactly as before.
+ final Headers headers = bufferedContext.headers();
+ key = keySerde.deserializer().deserialize(changelogTopic,
headers, next.getKey().key().get());
+ value = new Change<>(
+ deserializeValue(bufferValue.newValue(), headers),
+ deserializeValue(bufferValue.oldValue(), headers)
+ );
+ evictionContext = bufferedContext;
+ }
+ callback.accept(new Eviction<K, Change<V>>(key, value,
evictionContext));
delegate.remove();
index.remove(next.getKey().key());
@@ -434,15 +526,41 @@ public final class
InMemoryTimeOrderedKeyValueChangeBuffer<K, V, T> implements T
}
}
+ // Copy of the given context with different headers, preserving everything
else -- including
+ // sourceRawKey/sourceRawValue, which the shorter ProcessorRecordContext
constructor drops.
+ private static ProcessorRecordContext withHeaders(final
ProcessorRecordContext context, final Headers headers) {
+ if (context.headers() == headers) {
+ return context;
+ }
+ return new ProcessorRecordContext(
+ context.timestamp(),
+ context.offset(),
+ context.partition(),
+ context.topic(),
+ headers,
+ context.sourceRawKey(),
+ context.sourceRawValue()
+ );
+ }
+
@Override
public Maybe<ValueTimestampHeaders<V>> priorValueForBuffered(final K key) {
final Bytes serializedKey =
Bytes.wrap(keySerde.serializer().serialize(changelogTopic, context.headers(),
key));
- if (index.containsKey(serializedKey)) {
- final byte[] serializedValue =
internalPriorValueForBuffered(serializedKey);
+ final BufferKey bufferKey = index.get(serializedKey);
+ if (bufferKey != null) {
+ final BufferValue bufferValue = sortedMap.get(bufferKey);
+ final byte[] serializedValue = bufferValue.priorValue();
+
+ if (storeHeaders) {
+ // The prior value is stored as a ValueTimestampHeaders blob,
so we can recover its
+ // timestamp and headers directly (they are unknown/empty when
the key was first
+ // buffered, but preserved across restarts via the changelog).
+ return
Maybe.defined(deserializeValueTimestampHeaders(serializedValue));
+ }
final V deserializedValue =
valueSerde.innerSerde().deserializer().deserialize(
changelogTopic,
- context.headers(),
+ bufferValue.context().headers(),
serializedValue
);
@@ -465,6 +583,102 @@ public final class
InMemoryTimeOrderedKeyValueChangeBuffer<K, V, T> implements T
}
}
+ // 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.
The new value gets them
+ // from the restored record context, which describes exactly that part
(see put()); the prior and
+ // old values get them from the changelog record's own Kafka headers.
Records written before this
+ // feature -- or by a run configured without header stores -- carry no
such Kafka headers, in
+ // which case the prior/old 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 ProcessorRecordContext context = value.context();
+ final byte[] oldValue = valuePartWithHeaders(value.oldValue(),
recordHeaders, OLD_VALUE_HEADERS_KEY, context.timestamp());
+ // A missing vh.prior on a row whose prior and old values share an
array means the writer
+ // skipped the duplicate prefix, so the old part IS the prior part.
The two conditions have to
+ // be checked together: the changelog value dedups on the PLAIN bytes,
so a row can come back
+ // sharing an array even though the parts carried different headers
and were written with a
+ // vh.prior of their own. In that case the header wins.
+ final byte[] priorValue = value.priorValue() == value.oldValue()
+ && recordHeaders.lastHeader(PRIOR_VALUE_HEADERS_KEY) == null
+ ? oldValue
+ : valuePartWithHeaders(value.priorValue(), recordHeaders,
PRIOR_VALUE_HEADERS_KEY, context.timestamp());
+ return new BufferValue(
+ priorValue,
+ oldValue,
+ Utils.rawValueTimestampHeaders(value.newValue(),
context.timestamp(), context.headers()),
+ context
+ );
+ }
+
+ private static byte[] valuePartWithHeaders(final byte[] rawPlainValue,
+ final Headers recordHeaders,
+ final String headerKey,
+ final long fallbackTimestamp) {
+ if (rawPlainValue == null) {
+ return null;
+ }
+ final Header headersAndTimestamp = recordHeaders.lastHeader(headerKey);
+ return headersAndTimestamp == null
+ ? Utils.rawValueTimestampHeaders(rawPlainValue, fallbackTimestamp)
+ : Utils.rawValueTimestampHeaders(headersAndTimestamp.value(),
rawPlainValue);
+ }
+
+ 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> deserializeValueTimestampHeaders(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(deserializeValueTimestampHeaders(bytes));
+ }
+ return
valueSerde.innerSerde().deserializer().deserialize(changelogTopic,
fallbackHeaders, bytes);
+ }
+
@Override
public boolean put(final long time,
final Record<K, Change<V>> record,
@@ -472,21 +686,44 @@ public final class
InMemoryTimeOrderedKeyValueChangeBuffer<K, V, T> implements T
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 headers of the record being processed. These describe the new
value: the old value came
+ // from an earlier record and gets its own headers below. The
framework keeps recordContext in
+ // sync with the record (see StreamTask#doProcess and
ProcessorContextImpl#forward), so this is
+ // the same object as recordContext.headers(); we read it off the
record because that is where
+ // it conceptually belongs.
+ final Headers headers = 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 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 =
deserializeValueTimestampHeaders(buffered.newValue());
+ if (previousNewValue != null) {
+ oldHeaders = previousNewValue.headers();
+ oldTimestamp = previousNewValue.timestamp();
+ }
}
+ final Change<V> change = record.value();
+ // Order matters, and must stay old-then-new as in
FullChangeSerde#serializeParts: a serializer
+ // may write into the headers it is handed (Schema Registry serdes
record the schema id there),
+ // and in plain mode both parts are serialized against the same live
record headers, so
+ // whichever is serialized last determines what the emitted record
carries. That has to be the
+ // new value. In plain mode the old value is therefore serialized with
the current record's headers
+ // exactly as before, leaving the stored bytes unchanged for
non-header stores.
+ final byte[] oldValue = serializeValuePart(change.oldValue,
oldTimestamp, storeHeaders ? oldHeaders : headers);
+ final byte[] newValue = serializeValuePart(change.newValue, timestamp,
headers);
+ final byte[] serializedPriorValue = buffered == null ? oldValue :
buffered.priorValue();
+
cleanPut(
time,
serializedKey,
- new BufferValue(serializedPriorValue, serialChange.oldValue,
serialChange.newValue, recordContext)
+ new BufferValue(serializedPriorValue, oldValue, newValue,
recordContext)
);
if (loggingEnabled) {
dirtyKeys.add(serializedKey);
diff --git
a/streams/src/main/java/org/apache/kafka/streams/state/internals/Utils.java
b/streams/src/main/java/org/apache/kafka/streams/state/internals/Utils.java
index 6b9d3827587..6c899847641 100644
--- a/streams/src/main/java/org/apache/kafka/streams/state/internals/Utils.java
+++ b/streams/src/main/java/org/apache/kafka/streams/state/internals/Utils.java
@@ -107,6 +107,89 @@ public class Utils {
return result;
}
+ /**
+ * Build a serialized ValueTimestampHeaders with empty headers from a
plain value and a timestamp.
+ * This is the inverse of {@link #rawPlainValue(byte[])} for the
empty-headers case.
+ *
+ * Format conversion:
+ * Input: [value], timestamp
+ * Output: [headersSize(varint)=0][timestamp(8)][value]
+ */
+ public static byte[] rawValueTimestampHeaders(final byte[] rawPlainValue,
final long timestamp) {
+ return rawValueTimestampHeaders(rawPlainValue, timestamp, null);
+ }
+
+ /**
+ * Build a serialized ValueTimestampHeaders from a plain value, a
timestamp and headers.
+ * This is the inverse of {@link #rawPlainValue(byte[])}.
+ *
+ * Format conversion:
+ * Input: [value], timestamp, headers
+ * Output: [headersSize(varint)][headers][timestamp(8)][value]
+ */
+ public static byte[] rawValueTimestampHeaders(final byte[] rawPlainValue,
+ final long timestamp,
+ final Headers headers) {
+ if (rawPlainValue == null) {
+ return null;
+ }
+
+ final HeadersSerializer.PreSerializedHeaders preSerializedHeaders =
HeadersSerializer.prepareSerialization(headers);
+ final ByteBuffer buffer = ByteBuffer.allocate(
+
ByteUtils.sizeOfVarint(preSerializedHeaders.requiredBufferSizeForHeaders)
+ + preSerializedHeaders.requiredBufferSizeForHeaders
+ + StateSerdes.TIMESTAMP_SIZE
+ + rawPlainValue.length);
+
ByteUtils.writeVarint(preSerializedHeaders.requiredBufferSizeForHeaders,
buffer);
+ return HeadersSerializer.serialize(preSerializedHeaders, buffer)
+ .putLong(timestamp)
+ .put(rawPlainValue)
+ .array();
+ }
+
+ /**
+ * Extract the headers-and-timestamp prefix of a serialized
ValueTimestampHeaders, i.e. exactly
+ * the part that {@link #rawPlainValue(byte[])} strips off. Splitting a
value this way lets the
+ * plain value bytes and their headers/timestamp be stored or transmitted
separately and then
+ * recombined with {@link #rawValueTimestampHeaders(byte[], byte[])}.
+ *
+ * Format conversion:
+ * Input: [headersSize(varint)][headers][timestamp(8)][value]
+ * Output: [headersSize(varint)][headers][timestamp(8)]
+ */
+ public static byte[] rawHeadersTimestampPrefix(final byte[]
rawValueTimestampHeaders) {
+ if (rawValueTimestampHeaders == null) {
+ return null;
+ }
+
+ final ByteBuffer buffer = ByteBuffer.wrap(rawValueTimestampHeaders);
+ final int headersSize = ByteUtils.readVarint(buffer);
+ final int prefixLength = buffer.position() + headersSize +
StateSerdes.TIMESTAMP_SIZE;
+
+ final byte[] prefix = new byte[prefixLength];
+ System.arraycopy(rawValueTimestampHeaders, 0, prefix, 0, prefixLength);
+ return prefix;
+ }
+
+ /**
+ * Rebuild a serialized ValueTimestampHeaders from a prefix produced by
+ * {@link #rawHeadersTimestampPrefix(byte[])} and the plain value bytes it
was split from.
+ *
+ * Format conversion:
+ * Input: [headersSize(varint)][headers][timestamp(8)], [value]
+ * Output: [headersSize(varint)][headers][timestamp(8)][value]
+ */
+ public static byte[] rawValueTimestampHeaders(final byte[]
rawHeadersTimestampPrefix, final byte[] rawPlainValue) {
+ if (rawPlainValue == null) {
+ return null;
+ }
+
+ final byte[] result = new byte[rawHeadersTimestampPrefix.length +
rawPlainValue.length];
+ System.arraycopy(rawHeadersTimestampPrefix, 0, result, 0,
rawHeadersTimestampPrefix.length);
+ System.arraycopy(rawPlainValue, 0, result,
rawHeadersTimestampPrefix.length, rawPlainValue.length);
+ return result;
+ }
+
/**
* Extract raw timestamped value (timestamp + value) from serialized
ValueTimestampHeaders.
* This strips the headers portion but keeps timestamp and value intact.
diff --git
a/streams/src/test/java/org/apache/kafka/streams/kstream/internals/SuppressHeadersScenarioTest.java
b/streams/src/test/java/org/apache/kafka/streams/kstream/internals/SuppressHeadersScenarioTest.java
index d31d9880622..3f17d275c13 100644
---
a/streams/src/test/java/org/apache/kafka/streams/kstream/internals/SuppressHeadersScenarioTest.java
+++
b/streams/src/test/java/org/apache/kafka/streams/kstream/internals/SuppressHeadersScenarioTest.java
@@ -47,7 +47,6 @@ import org.apache.kafka.test.TestUtils;
import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.BeforeEach;
-import org.junit.jupiter.api.Disabled;
import org.junit.jupiter.api.Test;
import java.nio.charset.StandardCharsets;
@@ -115,10 +114,14 @@ import static org.junit.jupiter.api.Assertions.assertTrue;
* INV-1 is therefore unsatisfiable in plain format for these scenarios, no
matter how
* {@code suppress()} is implemented.
*
- * <p>{@link #nonWindowedEvictionTriggeredBySameKey()} and
- * {@link #windowedEvictionTriggeredBySameKeySameWindow()} consequently set
+ * <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.
+ * 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.
@@ -164,7 +167,6 @@ public class SuppressHeadersScenarioTest {
// ------------------------------------------------------------------
scenarios
- @Disabled("Enabled by the fix for KAFKA-20413; see class javadoc")
@Test
public void nonWindowedEvictionTriggeredByDifferentKey() {
scenario = "non-windowed / eviction triggered by a DIFFERENT key";
@@ -187,7 +189,6 @@ public class SuppressHeadersScenarioTest {
* The row holds {@code new=v2 / old=v1}, each from a different input
record, which plain values
* cannot represent.
*/
- @Disabled("Enabled by the fix for KAFKA-20413; see class javadoc")
@Test
public void nonWindowedEvictionTriggeredBySameKey() {
scenario = "non-windowed / SAME key, row updated then evicted (headers
format)";
@@ -218,10 +219,13 @@ public class SuppressHeadersScenarioTest {
* emitted; with a tombstone only the <em>old</em> value is serialized,
which makes the leak
* visible on the emitted record.
*/
- @Disabled("Enabled by the fix for KAFKA-20413; see class javadoc")
@Test
public void nonWindowedTombstoneMustBeForwarded() {
- scenario = "non-windowed / buffered row is DELETED, tombstone must be
forwarded";
+ scenario = "non-windowed / buffered row is DELETED, tombstone must be
forwarded (headers format)";
+ // Same-row overwrite: the delete lands on the row it deletes, so the
row still holds an old
+ // value from the earlier record. INV-1 therefore needs two origins
recoverable from one row;
+ // see the class javadoc on why that requires the HEADERS format.
+ config.setProperty(StreamsConfig.DSL_STORE_FORMAT_CONFIG,
StreamsConfig.DSL_STORE_FORMAT_HEADERS);
try (final TopologyTestDriver driver = new
TopologyTestDriverBuilder(nonWindowedTopology()).withConfig(config).build()) {
final TestInputTopic<String, String> input =
driver.createInputTopic(INPUT_TOPIC, new StringSerializer(),
new StringSerializer());
@@ -242,7 +246,6 @@ public class SuppressHeadersScenarioTest {
* it records: a <em>same</em> input key still produces a
<em>different</em> suppress row once
* windowed, so the evicted row is not the one the arriving record updated.
*/
- @Disabled("Enabled by the fix for KAFKA-20413; see class javadoc")
@Test
public void windowedEvictionTriggeredBySameKeyDifferentWindow() {
scenario = "windowed (untilWindowCloses) / SAME key but a DIFFERENT
window";
@@ -265,7 +268,6 @@ public class SuppressHeadersScenarioTest {
* under {@code untilTimeLimit}. It cannot happen under {@code
untilWindowCloses}, where being
* inside a window excludes being past its close.
*/
- @Disabled("Enabled by the fix for KAFKA-20413; see class javadoc")
@Test
public void windowedEvictionTriggeredBySameKeySameWindow() {
scenario = "windowed (untilTimeLimit) / SAME key and SAME window
(headers format)";
@@ -296,10 +298,12 @@ public class SuppressHeadersScenarioTest {
* <p>A null value cannot be piped into an aggregation, so the tombstone
is staged by having the
* reducer return {@code null} for the sentinel value {@code DELETE}.
*/
- @Disabled("Enabled by the fix for KAFKA-20413; see class javadoc")
@Test
public void windowedFinalResultsTombstoneIsDropped() {
- scenario = "windowed (untilWindowCloses) / tombstone is dropped by
design";
+ scenario = "windowed (untilWindowCloses) / tombstone is dropped by
design (headers format)";
+ // Same-row overwrite as above (kA's aggregate is updated to null), so
this too needs the
+ // HEADERS format for INV-1 to be satisfiable at all.
+ config.setProperty(StreamsConfig.DSL_STORE_FORMAT_CONFIG,
StreamsConfig.DSL_STORE_FORMAT_HEADERS);
try (final TopologyTestDriver driver = new
TopologyTestDriverBuilder(windowedTopology(false)).withConfig(config).build()) {
final TestInputTopic<String, String> input =
driver.createInputTopic(INPUT_TOPIC, new StringSerializer(),
new StringSerializer());
diff --git
a/streams/src/test/java/org/apache/kafka/streams/state/internals/TimeOrderedKeyValueBufferTest.java
b/streams/src/test/java/org/apache/kafka/streams/state/internals/TimeOrderedKeyValueBufferTest.java
index 3efbfdf5b53..f944ec23e81 100644
---
a/streams/src/test/java/org/apache/kafka/streams/state/internals/TimeOrderedKeyValueBufferTest.java
+++
b/streams/src/test/java/org/apache/kafka/streams/state/internals/TimeOrderedKeyValueBufferTest.java
@@ -19,12 +19,17 @@ package org.apache.kafka.streams.state.internals;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.producer.ProducerRecord;
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 org.apache.kafka.common.record.TimestampType;
+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.common.utils.Utils;
import org.apache.kafka.streams.KeyValue;
import org.apache.kafka.streams.StreamsConfig;
@@ -33,18 +38,21 @@ import org.apache.kafka.streams.processor.TaskId;
import org.apache.kafka.streams.processor.api.Record;
import org.apache.kafka.streams.processor.internals.ProcessorRecordContext;
import
org.apache.kafka.streams.processor.internals.RecordBatchingStateRestoreCallback;
+import org.apache.kafka.streams.processor.internals.RecordQueue;
import org.apache.kafka.streams.state.ValueTimestampHeaders;
import
org.apache.kafka.streams.state.internals.TimeOrderedKeyValueBuffer.Eviction;
import org.apache.kafka.test.MockInternalProcessorContext;
import org.apache.kafka.test.MockRecordCollector;
import org.apache.kafka.test.TestUtils;
+import org.junit.jupiter.api.Test;
import org.junit.jupiter.params.ParameterizedTest;
import org.junit.jupiter.params.provider.Arguments;
import org.junit.jupiter.params.provider.MethodSource;
import java.io.IOException;
import java.nio.ByteBuffer;
+import java.util.ArrayList;
import java.util.LinkedList;
import java.util.List;
import java.util.Map;
@@ -60,13 +68,20 @@ import static java.nio.charset.StandardCharsets.UTF_8;
import static java.util.Arrays.asList;
import static java.util.Collections.singletonList;
import static
org.apache.kafka.streams.state.internals.InMemoryTimeOrderedKeyValueChangeBuffer.CHANGELOG_HEADERS;
+import static
org.apache.kafka.streams.state.internals.InMemoryTimeOrderedKeyValueChangeBuffer.OLD_VALUE_HEADERS_KEY;
+import static
org.apache.kafka.streams.state.internals.InMemoryTimeOrderedKeyValueChangeBuffer.PRIOR_VALUE_HEADERS_KEY;
+import static
org.apache.kafka.streams.state.internals.Utils.rawValueTimestampHeaders;
import static org.hamcrest.MatcherAssert.assertThat;
import static org.hamcrest.Matchers.is;
+import static org.hamcrest.Matchers.not;
+import static org.hamcrest.Matchers.nullValue;
import static org.junit.jupiter.api.Assertions.fail;
public class TimeOrderedKeyValueBufferTest<B extends
TimeOrderedKeyValueBuffer<String, String, Change<String>>> {
private static final String APP_ID = "test-app";
+ /** Store name for the tests that build a buffer directly rather than
through {@link #parameters()}. */
+ private static final String STORE_NAME = "test-buffer";
private Function<String, B> bufferSupplier;
private String testName;
@@ -97,9 +112,16 @@ public class TimeOrderedKeyValueBufferTest<B extends
TimeOrderedKeyValueBuffer<S
}
private static MockInternalProcessorContext<?, ?> makeContext() {
+ return makeContext(false);
+ }
+
+ private static MockInternalProcessorContext<?, ?> makeContext(final
boolean headersEnabled) {
final Properties properties = new Properties();
properties.setProperty(StreamsConfig.APPLICATION_ID_CONFIG, APP_ID);
properties.setProperty(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG,
"mock:localhost:9092");
+ if (headersEnabled) {
+ properties.setProperty(StreamsConfig.DSL_STORE_FORMAT_CONFIG,
StreamsConfig.DSL_STORE_FORMAT_HEADERS);
+ }
final TaskId taskId = new TaskId(0, 0);
@@ -110,6 +132,19 @@ public class TimeOrderedKeyValueBufferTest<B extends
TimeOrderedKeyValueBuffer<S
}
+ /** Replays everything the source context's collector captured into the
restore context's callback. */
+ private static void restoreInto(final MockInternalProcessorContext<?, ?>
restoreContext,
+ final MockInternalProcessorContext<?, ?>
sourceContext,
+ final String storeName) {
+ final List<ConsumerRecord<byte[], byte[]>> toRestore = new
LinkedList<>();
+ for (final ProducerRecord<Object, Object> pr : ((MockRecordCollector)
sourceContext.recordCollector()).collected()) {
+ toRestore.add(new ConsumerRecord<>(
+ "changelog-topic", 0, 0, 999, TimestampType.CREATE_TIME, -1,
-1,
+ ((Bytes) pr.key()).get(), (byte[]) pr.value(), pr.headers(),
Optional.empty()));
+ }
+ ((RecordBatchingStateRestoreCallback)
restoreContext.stateRestoreCallback(storeName)).restoreBatch(toRestore);
+ }
+
private static void cleanup(final MockInternalProcessorContext<?, ?>
context, final TimeOrderedKeyValueBuffer<String, String, Change<String>>
buffer) {
try {
buffer.close();
@@ -310,6 +345,568 @@ public class TimeOrderedKeyValueBufferTest<B extends
TimeOrderedKeyValueBuffer<S
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))});
+ // The framework keeps the record context in sync with the record
being processed
+ // (StreamTask#doProcess, ProcessorContextImpl#forward), so both carry
the same headers here.
+ final ProcessorRecordContext recordContext = new
ProcessorRecordContext(0L, 0, 0, "topic", headers);
+ context.setRecordContext(recordContext);
+ buffer.put(0L, new Record<>("k", new Change<>("v", null), 0L,
headers), recordContext);
+
+ final List<Eviction<String, Change<String>>> evicted = new
LinkedList<>();
+ buffer.evictWhile(() -> true, evicted::add);
+
+ assertThat(evicted.size(), is(1));
+ assertThat(evicted.get(0).recordContext().headers(), is(headers));
+ cleanup(context, buffer);
+ }
+
+ @Test
+ public void
shouldDeserializeEachValuePartWithItsOwnHeadersWhenHeadersEnabled() {
+ // In headers mode the old and the new value of a buffered row
originate from two different
+ // input records, so each must be handed the headers of the record it
came from. A
+ // header-dependent deserializer (as e.g. Schema Registry serdes are,
and String serdes are
+ // not) records which headers it actually sees.
+ final List<String> headerSeenByDeserializer = new ArrayList<>();
+ final Deserializer<String> recordingDeserializer = new
Deserializer<>() {
+ @Override
+ public String deserialize(final String topic, final byte[] data) {
+ return data == null ? null : new String(data, UTF_8);
+ }
+
+ @Override
+ public String deserialize(final String topic, final Headers
headers, final byte[] data) {
+ final Header header = headers.lastHeader("h");
+ headerSeenByDeserializer.add(header == null ? "none" : new
String(header.value(), UTF_8));
+ return deserialize(topic, data);
+ }
+ };
+
+ final InMemoryTimeOrderedKeyValueChangeBuffer<String, String,
Change<String>> buffer =
+ new InMemoryTimeOrderedKeyValueChangeBuffer.Builder<>(
+ "test-buffer", Serdes.String(), Serdes.serdeFrom(new
StringSerializer(), recordingDeserializer)).build();
+ final MockInternalProcessorContext<?, ?> context = makeContext(true);
+ buffer.init(context, buffer);
+
+ final RecordHeaders headersA = new RecordHeaders(new Header[]{new
RecordHeader("h", "A".getBytes(UTF_8))});
+ final RecordHeaders headersB = new RecordHeaders(new Header[]{new
RecordHeader("h", "B".getBytes(UTF_8))});
+
+ // Record 1 (headers A) first buffers "k"="v1".
+ final ProcessorRecordContext contextA = new
ProcessorRecordContext(10L, 0, 0, "topic", headersA);
+ context.setRecordContext(contextA);
+ buffer.put(0L, new Record<>("k", new Change<>("v1", null), 10L,
headersA), contextA);
+
+ // Record 2 (headers B) updates "k" in place, so "v1" becomes the old
value of the row while
+ // the new value "v2" belongs to record 2.
+ final ProcessorRecordContext contextB = new
ProcessorRecordContext(20L, 1, 0, "topic", headersB);
+ context.setRecordContext(contextB);
+ buffer.put(0L, new Record<>("k", new Change<>("v2", "v1"), 20L,
headersB), contextB);
+
+ // The second put reads back the previous new value to recover the old
value's headers; only
+ // the eviction is under test here.
+ headerSeenByDeserializer.clear();
+
+ final List<Eviction<String, Change<String>>> evicted = new
LinkedList<>();
+ buffer.evictWhile(() -> true, evicted::add);
+
+ assertThat(evicted.size(), is(1));
+ assertThat(evicted.get(0).value(), is(new Change<>("v2", "v1")));
+ // New value first with its own headers (B), then the old value with
the headers of the record
+ // it originally arrived on (A) -- not with B, and not with whatever
triggered the eviction.
+ assertThat(headerSeenByDeserializer, is(List.of("B", "A")));
+ // The emitted record carries the new value's headers.
+ assertThat(evicted.get(0).recordContext().headers(), is(headersB));
+ cleanup(context, buffer);
+ }
+
+ @Test
+ public void
shouldDeserializeEvictedValueWithBufferedHeadersNotEvictionTriggerHeaders() {
+ // A header-dependent value deserializer (as e.g. Schema Registry
serdes are, and String
+ // serdes are not) records which headers it is handed. This lets us
prove that on eviction the
+ // buffered value is deserialized with the headers it was buffered
with, and not with the
+ // headers of whatever record happened to trigger the eviction.
+ final List<String> headerSeenByDeserializer = new ArrayList<>();
+ final Deserializer<String> recordingDeserializer = new
Deserializer<>() {
+ @Override
+ public String deserialize(final String topic, final byte[] data) {
+ return data == null ? null : new String(data, UTF_8);
+ }
+
+ @Override
+ public String deserialize(final String topic, final Headers
headers, final byte[] data) {
+ final Header header = headers.lastHeader("h");
+ headerSeenByDeserializer.add(header == null ? "none" : new
String(header.value(), UTF_8));
+ return deserialize(topic, data);
+ }
+ };
+ final Serde<String> valueSerde = Serdes.serdeFrom(new
StringSerializer(), recordingDeserializer);
+
+ final InMemoryTimeOrderedKeyValueChangeBuffer<String, String,
Change<String>> buffer =
+ new
InMemoryTimeOrderedKeyValueChangeBuffer.Builder<>("test-buffer",
Serdes.String(), valueSerde).build();
+ final MockInternalProcessorContext<?, ?> context = makeContext();
+ buffer.init(context, buffer);
+
+ // Buffer key "k" while the processing context carries header h=A.
+ final RecordHeaders bufferedHeaders = new RecordHeaders(new
Header[]{new RecordHeader("h", "A".getBytes(UTF_8))});
+ final ProcessorRecordContext bufferedContext = new
ProcessorRecordContext(0L, 0, 0, "topic", bufferedHeaders);
+ context.setRecordContext(bufferedContext);
+ buffer.put(0L, new Record<>("k", new Change<>("v", null), 0L,
bufferedHeaders), bufferedContext);
+
+ // Eviction happens later, while a DIFFERENT record (header h=B) is
being processed.
+ context.setRecordContext(new ProcessorRecordContext(1L, 1, 0, "topic",
+ new RecordHeaders(new Header[]{new RecordHeader("h",
"B".getBytes(UTF_8))})));
+
+ final List<Eviction<String, Change<String>>> evicted = new
LinkedList<>();
+ buffer.evictWhile(() -> true, evicted::add);
+
+ assertThat(evicted.size(), is(1));
+ // The buffered value must be deserialized with its own headers ("A"),
not the headers of the
+ // record that triggered the eviction ("B").
+ assertThat(headerSeenByDeserializer, is(singletonList("A")));
+ cleanup(context, buffer);
+ }
+
+ @Test
+ public void shouldSerializeNewValueLastSoItsHeadersWin() {
+ // A serializer may write into the headers it is handed (Schema
Registry serdes record the
+ // schema id there). In plain mode both value parts are serialized
against the same live record
+ // headers, so the part serialized LAST determines what the emitted
record carries -- and that
+ // has to be the new value. This is why FullChangeSerde#serializeParts
serializes old before new.
+ final Serializer<String> headerWritingSerializer = new Serializer<>() {
+ @Override
+ public byte[] serialize(final String topic, final String data) {
+ return data == null ? null : data.getBytes(UTF_8);
+ }
+
+ @Override
+ public byte[] serialize(final String topic, final Headers headers,
final String data) {
+ headers.add(new RecordHeader("serialized",
data.getBytes(UTF_8)));
+ return serialize(topic, data);
+ }
+ };
+ final Serde<String> valueSerde =
Serdes.serdeFrom(headerWritingSerializer, new StringDeserializer());
+
+ final InMemoryTimeOrderedKeyValueChangeBuffer<String, String,
Change<String>> buffer =
+ new
InMemoryTimeOrderedKeyValueChangeBuffer.Builder<>("test-buffer",
Serdes.String(), valueSerde).build();
+ final MockInternalProcessorContext<?, ?> context = makeContext();
+ buffer.init(context, buffer);
+
+ // Record's constructor copies the headers it is given, so build the
context from the record's
+ // own headers object -- that is the object the framework ends up
sharing between the two
+ // (ProcessorContextImpl#forward re-points the context at
record.headers()), and the one that
+ // gets forwarded downstream.
+ final Record<String, Change<String>> record =
+ new Record<>("k", new Change<>("new", "old"), 0L, new
RecordHeaders());
+ final ProcessorRecordContext recordContext = new
ProcessorRecordContext(0L, 0, 0, "topic", record.headers());
+ context.setRecordContext(recordContext);
+ buffer.put(0L, record, recordContext);
+
+ assertThat(new
String(record.headers().lastHeader("serialized").value(), UTF_8), is("new"));
+
+ // A tombstone has no new value to serialize, so the old value's
header is the one that stands.
+ final Record<String, Change<String>> tombstone =
+ new Record<>("k2", new Change<>(null, "old"), 1L, new
RecordHeaders());
+ final ProcessorRecordContext tombstoneContext = new
ProcessorRecordContext(1L, 1, 0, "topic", tombstone.headers());
+ context.setRecordContext(tombstoneContext);
+ buffer.put(0L, tombstone, tombstoneContext);
+
+ assertThat(new
String(tombstone.headers().lastHeader("serialized").value(), UTF_8), is("old"));
+ cleanup(context, buffer);
+ }
+
+ @Test
+ public void shouldStorePerValueHeadersInChangelogWhenHeadersEnabled() {
+ // With dsl.store.format=HEADERS the old and new value parts each
carry their OWN headers and
+ // timestamp. Those must NOT go into the changelog value -- that has
to stay in the V3 format
+ // older versions can restore -- so they travel in the changelog
record's Kafka headers.
+ final InMemoryTimeOrderedKeyValueChangeBuffer<String, String,
Change<String>> buffer =
+ new
InMemoryTimeOrderedKeyValueChangeBuffer.Builder<>("test-buffer",
Serdes.String(), Serdes.String()).build();
+ final MockInternalProcessorContext<?, ?> context = makeContext(true);
+ buffer.init(context, buffer);
+
+ final RecordHeaders headersA = new RecordHeaders(new Header[]{new
RecordHeader("h", "A".getBytes(UTF_8))});
+ final RecordHeaders headersB = new RecordHeaders(new Header[]{new
RecordHeader("h", "B".getBytes(UTF_8))});
+
+ // First buffer "k" (value "v1") with headers A at record timestamp 10.
+ final ProcessorRecordContext contextA = new
ProcessorRecordContext(10L, 0, 0, "topic", headersA);
+ context.setRecordContext(contextA);
+ buffer.put(0L, new Record<>("k", new Change<>("v1", null), 10L,
headersA), contextA);
+
+ // In-place update (value "v2", old "v1") with headers B at record
timestamp 20. The old value
+ // ("v1") should keep the first record's headers/timestamp (A / 10)
via carry-forward.
+ final ProcessorRecordContext contextB = new
ProcessorRecordContext(20L, 1, 0, "topic", headersB);
+ context.setRecordContext(contextB);
+ buffer.put(0L, new Record<>("k", new Change<>("v2", "v1"), 20L,
headersB), contextB);
+
+ buffer.commit(Map.of());
+
+ final List<ProducerRecord<Object, Object>> collected =
((MockRecordCollector) context.recordCollector()).collected();
+ assertThat(collected.size(), is(1));
+ final ProducerRecord<Object, Object> changelogRecord =
collected.get(0);
+
+ // The version marker stays at V3, so an older version restores this
record fine...
+ assertThat(changelogRecord.headers().lastHeader("v").value(), is(new
byte[] {(byte) 3}));
+
+ // ...because the value bytes are plain values, exactly as the V3
format prescribes.
+ final BufferValue bufferValue =
BufferValue.deserialize(ByteBuffer.wrap((byte[]) changelogRecord.value()));
+ final StringDeserializer plainDeserializer = new StringDeserializer();
+ assertThat(plainDeserializer.deserialize("topic",
bufferValue.newValue()), is("v2"));
+ assertThat(plainDeserializer.deserialize("topic",
bufferValue.oldValue()), is("v1"));
+
+ // The new value's headers and timestamp need no Kafka header of their
own: the record context
+ // encoded in the V3 value already describes that part, since it is
the context of the very
+ // record the new value came from.
+ assertThat(bufferValue.context().headers(), is(headersB));
+ assertThat(bufferValue.context().timestamp(), is(20L));
+
+ // The prior and old parts have no such carrier, so their headers and
timestamps ride in the
+ // record's Kafka headers, and recombine with the plain value bytes
into the in-memory encoding.
+ final ValueTimestampHeadersDeserializer<String> deserializer =
+ new ValueTimestampHeadersDeserializer<>(new StringDeserializer());
+
+ final ValueTimestampHeaders<String> oldValue =
deserializer.deserialize("topic",
+
rawValueTimestampHeaders(changelogRecord.headers().lastHeader(OLD_VALUE_HEADERS_KEY).value(),
bufferValue.oldValue()));
+ assertThat(oldValue.value(), is("v1"));
+ assertThat(oldValue.timestamp(), is(10L)); // carried forward from
the first record
+ assertThat(oldValue.headers(), is(headersA)); // carried forward from
the first record
+
+ cleanup(context, buffer);
+ }
+
+ @Test
+ public void
shouldNotWritePriorValueHeadersWhenPriorAndOldValueShareAnArray() {
+ // On the first buffering of a key the prior value IS the old value:
BufferValue collapses them
+ // onto one array and the V3 serialization writes those bytes only
once. The per-part headers
+ // must not undo that saving by writing the same prefix under a second
key.
+ final InMemoryTimeOrderedKeyValueChangeBuffer<String, String,
Change<String>> buffer =
+ new InMemoryTimeOrderedKeyValueChangeBuffer.Builder<>(STORE_NAME,
Serdes.String(), Serdes.String()).build();
+ final MockInternalProcessorContext<?, ?> context = makeContext(true);
+ buffer.init(context, buffer);
+
+ final RecordHeaders headers = new RecordHeaders(new Header[]{new
RecordHeader("h", "A".getBytes(UTF_8))});
+ final ProcessorRecordContext recordContext = new
ProcessorRecordContext(10L, 0, 0, "topic", headers);
+ context.setRecordContext(recordContext);
+ buffer.put(0L, new Record<>("k", new Change<>("v1", "p"), 10L,
headers), recordContext);
+ buffer.commit(Map.of());
+
+ final ProducerRecord<Object, Object> changelogRecord =
+ ((MockRecordCollector)
context.recordCollector()).collected().get(0);
+
assertThat(changelogRecord.headers().lastHeader(OLD_VALUE_HEADERS_KEY),
is(not(nullValue())));
+
assertThat(changelogRecord.headers().lastHeader(PRIOR_VALUE_HEADERS_KEY),
is(nullValue()));
+
+ // The prior value must still come back, recovered from the old part
rather than from the
+ // header the writer deliberately left out.
+ final InMemoryTimeOrderedKeyValueChangeBuffer<String, String,
Change<String>> restored =
+ new InMemoryTimeOrderedKeyValueChangeBuffer.Builder<>(STORE_NAME,
Serdes.String(), Serdes.String()).build();
+ final MockInternalProcessorContext<?, ?> restoreContext =
makeContext(true);
+ restored.init(restoreContext, restored);
+ restoreInto(restoreContext, context, STORE_NAME);
+
+ assertThat(restored.priorValueForBuffered("k"),
+ is(Maybe.defined(ValueTimestampHeaders.make("p",
RecordQueue.UNKNOWN, new RecordHeaders()))));
+ cleanup(restoreContext, restored);
+ cleanup(context, buffer);
+ }
+
+ @Test
+ public void
shouldKeepPriorValueHeadersWhenOnlyThePlainBytesOfPriorAndOldValueMatch() {
+ // The counter-case that makes the dedup above non-trivial: the
changelog value dedups on the
+ // PLAIN bytes, so a row can come back from the changelog sharing an
array even though its prior
+ // and old parts carried different headers and timestamps. Here the
old value is written a second
+ // time with the same value bytes ("p") but picks up the first
record's headers by carry-forward,
+ // while the prior value keeps the unknown/empty origin it was first
buffered with.
+ final InMemoryTimeOrderedKeyValueChangeBuffer<String, String,
Change<String>> buffer =
+ new InMemoryTimeOrderedKeyValueChangeBuffer.Builder<>(STORE_NAME,
Serdes.String(), Serdes.String()).build();
+ final MockInternalProcessorContext<?, ?> context = makeContext(true);
+ buffer.init(context, buffer);
+
+ final RecordHeaders headersA = new RecordHeaders(new Header[]{new
RecordHeader("h", "A".getBytes(UTF_8))});
+ final RecordHeaders headersB = new RecordHeaders(new Header[]{new
RecordHeader("h", "B".getBytes(UTF_8))});
+
+ final ProcessorRecordContext contextA = new
ProcessorRecordContext(10L, 0, 0, "topic", headersA);
+ context.setRecordContext(contextA);
+ buffer.put(0L, new Record<>("k", new Change<>("v1", "p"), 10L,
headersA), contextA);
+
+ final ProcessorRecordContext contextB = new
ProcessorRecordContext(20L, 1, 0, "topic", headersB);
+ context.setRecordContext(contextB);
+ buffer.put(0L, new Record<>("k", new Change<>("v2", "p"), 20L,
headersB), contextB);
+
+ buffer.commit(Map.of());
+
+ // The parts differ, so this time the prefix must be written under its
own key.
+ final ProducerRecord<Object, Object> changelogRecord =
+ ((MockRecordCollector)
context.recordCollector()).collected().get(0);
+
assertThat(changelogRecord.headers().lastHeader(PRIOR_VALUE_HEADERS_KEY),
is(not(nullValue())));
+
+ final InMemoryTimeOrderedKeyValueChangeBuffer<String, String,
Change<String>> restored =
+ new InMemoryTimeOrderedKeyValueChangeBuffer.Builder<>(STORE_NAME,
Serdes.String(), Serdes.String()).build();
+ final MockInternalProcessorContext<?, ?> restoreContext =
makeContext(true);
+ restored.init(restoreContext, restored);
+ restoreInto(restoreContext, context, STORE_NAME);
+
+ // Not (p, 10, A) -- that is the OLD part, and taking it here would be
the dedup misfiring.
+ assertThat(restored.priorValueForBuffered("k"),
+ is(Maybe.defined(ValueTimestampHeaders.make("p",
RecordQueue.UNKNOWN, new RecordHeaders()))));
+ cleanup(restoreContext, restored);
+ cleanup(context, buffer);
+ }
+
+ @ParameterizedTest
+ @MethodSource("parameters")
+ public void
shouldPreservePriorValueTimestampAndHeadersWhenHeadersEnabled(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(true);
+ buffer.init(context, buffer);
+
+ final RecordHeaders headers = new RecordHeaders(new Header[]{new
RecordHeader("h1", "v1".getBytes(UTF_8))});
+ final ProcessorRecordContext recordContext = getContext(0L);
+ context.setRecordContext(recordContext);
+ buffer.put(1L, new Record<>("A", new Change<>("new-value",
"old-value"), 0L, headers), recordContext);
+ buffer.put(1L, new Record<>("B", new Change<>("new-value", null), 0L,
headers), recordContext);
+
+ // The prior value's original timestamp/headers are unknown when a key
is first buffered, so
+ // they round-trip through the ValueTimestampHeaders encoding as
UNKNOWN/empty.
+ assertThat(buffer.priorValueForBuffered("A"),
is(Maybe.defined(ValueTimestampHeaders.make("old-value", -1, new
RecordHeaders()))));
+ assertThat(buffer.priorValueForBuffered("B"), is(Maybe.defined(null)));
+ cleanup(context, buffer);
+ }
+
+ @ParameterizedTest
+ @MethodSource("parameters")
+ public void
shouldRoundTripHeadersThroughCommitAndRestoreWhenHeadersEnabled(final String
testName, final Function<String, B> bufferSupplier) {
+ setup(testName, bufferSupplier);
+
+ // Buffer a record (with headers and a record timestamp distinct from
the buffer time) and
+ // commit it to the changelog.
+ final TimeOrderedKeyValueBuffer<String, String, Change<String>> buffer
= bufferSupplier.apply(testName);
+ final MockInternalProcessorContext<?, ?> context = makeContext(true);
+ buffer.init(context, buffer);
+
+ final RecordHeaders headers = new RecordHeaders(new Header[]{new
RecordHeader("h1", "v1".getBytes(UTF_8))});
+ context.setRecordContext(new ProcessorRecordContext(5L, 0, 0, "topic",
headers));
+ buffer.put(0L, new Record<>("k", new Change<>("new", "old"), 5L,
headers), context.recordContext());
+ buffer.commit(Map.of());
+
+ final List<ProducerRecord<Object, Object>> collected =
((MockRecordCollector) context.recordCollector()).collected();
+ assertThat(collected.size(), is(1));
+
+ // Restore the changelog into a fresh buffer and confirm the value,
record timestamp and
+ // headers all survived the serialization round-trip.
+ final TimeOrderedKeyValueBuffer<String, String, Change<String>>
restored = bufferSupplier.apply(testName);
+ final MockInternalProcessorContext<?, ?> restoreContext =
makeContext(true);
+ restored.init(restoreContext, restored);
+ final RecordBatchingStateRestoreCallback stateRestoreCallback =
+ (RecordBatchingStateRestoreCallback)
restoreContext.stateRestoreCallback(testName);
+
+ final List<ConsumerRecord<byte[], byte[]>> toRestore = new
LinkedList<>();
+ for (final ProducerRecord<Object, Object> pr : collected) {
+ toRestore.add(new ConsumerRecord<>(
+ "changelog-topic", 0, 0, 999, TimestampType.CREATE_TIME, -1,
-1,
+ ((Bytes) pr.key()).get(), (byte[]) pr.value(), pr.headers(),
Optional.empty()));
+ }
+ stateRestoreCallback.restoreBatch(toRestore);
+
+ final List<Eviction<String, Change<String>>> evicted = new
LinkedList<>();
+ restored.evictWhile(() -> true, evicted::add);
+
+ assertThat(evicted.size(), is(1));
+ assertThat(evicted.get(0).key(), is("k"));
+ assertThat(evicted.get(0).value(), is(new Change<>("new", "old")));
+ assertThat(evicted.get(0).recordContext().timestamp(), is(5L));
+ assertThat(evicted.get(0).recordContext().headers(), is(headers));
+ cleanup(restoreContext, restored);
+ cleanup(context, buffer);
+ }
+
+ @Test
+ public void
shouldRoundTripDistinctPerValuePartHeadersThroughCommitAndRestore() {
+ // The point of the per-part changelog headers: a row whose old and
new value come from
+ // different records must come back from the changelog with BOTH
origins intact, not just one.
+ // A recording deserializer shows what each part is actually handed
after the restore.
+ final List<String> headerSeenByDeserializer = new ArrayList<>();
+ final Deserializer<String> recordingDeserializer = new
Deserializer<>() {
+ @Override
+ public String deserialize(final String topic, final byte[] data) {
+ return data == null ? null : new String(data, UTF_8);
+ }
+
+ @Override
+ public String deserialize(final String topic, final Headers
headers, final byte[] data) {
+ final Header header = headers.lastHeader("h");
+ headerSeenByDeserializer.add(header == null ? "none" : new
String(header.value(), UTF_8));
+ return deserialize(topic, data);
+ }
+ };
+ final Serde<String> valueSerde = Serdes.serdeFrom(new
StringSerializer(), recordingDeserializer);
+
+ final InMemoryTimeOrderedKeyValueChangeBuffer<String, String,
Change<String>> buffer =
+ new InMemoryTimeOrderedKeyValueChangeBuffer.Builder<>(STORE_NAME,
Serdes.String(), valueSerde).build();
+ final MockInternalProcessorContext<?, ?> context = makeContext(true);
+ buffer.init(context, buffer);
+
+ final RecordHeaders headersA = new RecordHeaders(new Header[]{new
RecordHeader("h", "A".getBytes(UTF_8))});
+ final RecordHeaders headersB = new RecordHeaders(new Header[]{new
RecordHeader("h", "B".getBytes(UTF_8))});
+
+ final ProcessorRecordContext contextA = new
ProcessorRecordContext(10L, 0, 0, "topic", headersA);
+ context.setRecordContext(contextA);
+ buffer.put(0L, new Record<>("k", new Change<>("v1", null), 10L,
headersA), contextA);
+
+ final ProcessorRecordContext contextB = new
ProcessorRecordContext(20L, 1, 0, "topic", headersB);
+ context.setRecordContext(contextB);
+ buffer.put(0L, new Record<>("k", new Change<>("v2", "v1"), 20L,
headersB), contextB);
+
+ buffer.commit(Map.of());
+
+ final InMemoryTimeOrderedKeyValueChangeBuffer<String, String,
Change<String>> restored =
+ new InMemoryTimeOrderedKeyValueChangeBuffer.Builder<>(STORE_NAME,
Serdes.String(), valueSerde).build();
+ final MockInternalProcessorContext<?, ?> restoreContext =
makeContext(true);
+ restored.init(restoreContext, restored);
+ restoreInto(restoreContext, context, STORE_NAME);
+
+ // Only the eviction of the restored buffer is under test.
+ headerSeenByDeserializer.clear();
+
+ final List<Eviction<String, Change<String>>> evicted = new
LinkedList<>();
+ restored.evictWhile(() -> true, evicted::add);
+
+ assertThat(evicted.size(), is(1));
+ assertThat(evicted.get(0).value(), is(new Change<>("v2", "v1")));
+ // New value with its own headers (B), then the old value with the
headers of the record it
+ // originally arrived on (A) -- both recovered from the changelog, not
just the latest one.
+ assertThat(headerSeenByDeserializer, is(List.of("B", "A")));
+ assertThat(evicted.get(0).recordContext().headers(), is(headersB));
+ cleanup(restoreContext, restored);
+ cleanup(context, buffer);
+ }
+
+ @ParameterizedTest
+ @MethodSource("parameters")
+ public void
shouldRestoreChangelogWrittenWithoutHeadersIntoBufferWithHeaders(final String
testName, final Function<String, B> bufferSupplier) {
+ setup(testName, bufferSupplier);
+
+ // The upgrade path, and the mirror of the downgrade test below: a
changelog written before
+ // this feature (or by a run without dsl.store.format=HEADERS) carries
no per-part headers.
+ // The new value still recovers its own headers and timestamp from the
encoded record context,
+ // but the prior and old originals are genuinely unknown and fall back
to empty headers with
+ // the record-context timestamp. The values themselves must still
restore intact.
+ final TimeOrderedKeyValueBuffer<String, String, Change<String>> buffer
= bufferSupplier.apply(testName);
+ final MockInternalProcessorContext<?, ?> context = makeContext(false);
+ buffer.init(context, buffer);
+
+ final RecordHeaders headers = new RecordHeaders(new Header[]{new
RecordHeader("h1", "v1".getBytes(UTF_8))});
+ context.setRecordContext(new ProcessorRecordContext(5L, 0, 0, "topic",
headers));
+ buffer.put(0L, new Record<>("k", new Change<>("new", "old"), 5L,
headers), context.recordContext());
+ buffer.commit(Map.of());
+
+ final TimeOrderedKeyValueBuffer<String, String, Change<String>>
restored = bufferSupplier.apply(testName);
+ final MockInternalProcessorContext<?, ?> restoreContext =
makeContext(true);
+ restored.init(restoreContext, restored);
+ restoreInto(restoreContext, context, testName);
+
+ // The prior value has no per-part headers to recover, so it falls
back to empty headers and
+ // the record-context timestamp rather than the UNKNOWN it would carry
in a headers changelog.
+ assertThat(restored.priorValueForBuffered("k"),
is(Maybe.defined(ValueTimestampHeaders.make("old", 5L, new RecordHeaders()))));
+
+ final List<Eviction<String, Change<String>>> evicted = new
LinkedList<>();
+ restored.evictWhile(() -> true, evicted::add);
+
+ assertThat(evicted.size(), is(1));
+ assertThat(evicted.get(0).key(), is("k"));
+ assertThat(evicted.get(0).value(), is(new Change<>("new", "old")));
+ assertThat(evicted.get(0).recordContext().timestamp(), is(5L));
+ cleanup(restoreContext, restored);
+ cleanup(context, buffer);
+ }
+
+ @ParameterizedTest
+ @MethodSource("parameters")
+ public void
shouldPreservePriorValueTimestampAndHeadersAcrossRestoreWhenHeadersEnabled(final
String testName, final Function<String, B> bufferSupplier) {
+ setup(testName, bufferSupplier);
+
+ // vh.prior exists so the prior value's headers and timestamp survive
the changelog; that value
+ // is surfaced to downstream value getters, so it has to come back
exactly as it went in. On a
+ // first insert they are genuinely unknown, which must round-trip as
UNKNOWN/empty rather than
+ // silently becoming the record-context timestamp.
+ final TimeOrderedKeyValueBuffer<String, String, Change<String>> buffer
= bufferSupplier.apply(testName);
+ final MockInternalProcessorContext<?, ?> context = makeContext(true);
+ buffer.init(context, buffer);
+
+ final RecordHeaders headers = new RecordHeaders(new Header[]{new
RecordHeader("h1", "v1".getBytes(UTF_8))});
+ context.setRecordContext(new ProcessorRecordContext(5L, 0, 0, "topic",
headers));
+ buffer.put(0L, new Record<>("k", new Change<>("new", "old"), 5L,
headers), context.recordContext());
+ buffer.commit(Map.of());
+
+ assertThat(buffer.priorValueForBuffered("k"),
+ is(Maybe.defined(ValueTimestampHeaders.make("old",
RecordQueue.UNKNOWN, new RecordHeaders()))));
+
+ final TimeOrderedKeyValueBuffer<String, String, Change<String>>
restored = bufferSupplier.apply(testName);
+ final MockInternalProcessorContext<?, ?> restoreContext =
makeContext(true);
+ restored.init(restoreContext, restored);
+ restoreInto(restoreContext, context, testName);
+
+ assertThat(restored.priorValueForBuffered("k"),
+ is(Maybe.defined(ValueTimestampHeaders.make("old",
RecordQueue.UNKNOWN, new RecordHeaders()))));
+ cleanup(restoreContext, restored);
+ cleanup(context, buffer);
+ }
+
+ @ParameterizedTest
+ @MethodSource("parameters")
+ public void
shouldRestoreChangelogWrittenWithHeadersIntoBufferWithoutHeaders(final String
testName, final Function<String, B> bufferSupplier) {
+ setup(testName, bufferSupplier);
+
+ // Offline downgrade: a changelog written by a run with
dsl.store.format=HEADERS must remain
+ // readable by a run without it (and, by the same token, by an older
version that knows
+ // nothing about the per-value-part record headers). This works
because the value bytes are
+ // plain V3 and the extra headers are simply ignored.
+ final TimeOrderedKeyValueBuffer<String, String, Change<String>> buffer
= bufferSupplier.apply(testName);
+ final MockInternalProcessorContext<?, ?> context = makeContext(true);
+ buffer.init(context, buffer);
+
+ final RecordHeaders headers = new RecordHeaders(new Header[]{new
RecordHeader("h1", "v1".getBytes(UTF_8))});
+ context.setRecordContext(new ProcessorRecordContext(5L, 0, 0, "topic",
headers));
+ buffer.put(0L, new Record<>("k", new Change<>("new", "old"), 5L,
headers), context.recordContext());
+ buffer.commit(Map.of());
+
+ final List<ProducerRecord<Object, Object>> collected =
((MockRecordCollector) context.recordCollector()).collected();
+ assertThat(collected.size(), is(1));
+
+ // Restore into a buffer configured WITHOUT header stores.
+ final TimeOrderedKeyValueBuffer<String, String, Change<String>>
restored = bufferSupplier.apply(testName);
+ final MockInternalProcessorContext<?, ?> restoreContext =
makeContext(false);
+ restored.init(restoreContext, restored);
+ final RecordBatchingStateRestoreCallback stateRestoreCallback =
+ (RecordBatchingStateRestoreCallback)
restoreContext.stateRestoreCallback(testName);
+
+ final List<ConsumerRecord<byte[], byte[]>> toRestore = new
LinkedList<>();
+ for (final ProducerRecord<Object, Object> pr : collected) {
+ toRestore.add(new ConsumerRecord<>(
+ "changelog-topic", 0, 0, 999, TimestampType.CREATE_TIME, -1,
-1,
+ ((Bytes) pr.key()).get(), (byte[]) pr.value(), pr.headers(),
Optional.empty()));
+ }
+ stateRestoreCallback.restoreBatch(toRestore);
+
+ final List<Eviction<String, Change<String>>> evicted = new
LinkedList<>();
+ restored.evictWhile(() -> true, evicted::add);
+
+ // The values come back intact; only the per-part headers are absent,
which is exactly what
+ // running without a headers store format means.
+ assertThat(evicted.size(), is(1));
+ assertThat(evicted.get(0).key(), is("k"));
+ assertThat(evicted.get(0).value(), is(new Change<>("new", "old")));
+ assertThat(evicted.get(0).recordContext().timestamp(), is(5L));
+ cleanup(restoreContext, restored);
+ cleanup(context, buffer);
+ }
+
@ParameterizedTest
@MethodSource("parameters")
public void shouldCommit(final String testName, final Function<String, B>
bufferSupplier) {
diff --git
a/streams/src/test/java/org/apache/kafka/streams/state/internals/UtilsTest.java
b/streams/src/test/java/org/apache/kafka/streams/state/internals/UtilsTest.java
index 1d08e898996..d1251936663 100644
---
a/streams/src/test/java/org/apache/kafka/streams/state/internals/UtilsTest.java
+++
b/streams/src/test/java/org/apache/kafka/streams/state/internals/UtilsTest.java
@@ -17,7 +17,9 @@
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.header.internals.RecordHeader;
import org.apache.kafka.common.header.internals.RecordHeaders;
import org.apache.kafka.common.serialization.Serdes;
import org.apache.kafka.common.serialization.StringSerializer;
@@ -37,8 +39,10 @@ import java.nio.charset.StandardCharsets;
import java.util.stream.Stream;
import static org.apache.kafka.streams.state.internals.Utils.hasEmptyHeaders;
+import static
org.apache.kafka.streams.state.internals.Utils.rawHeadersTimestampPrefix;
import static org.apache.kafka.streams.state.internals.Utils.rawPlainValue;
import static
org.apache.kafka.streams.state.internals.Utils.rawTimestampedValue;
+import static
org.apache.kafka.streams.state.internals.Utils.rawValueTimestampHeaders;
import static org.apache.kafka.streams.state.internals.Utils.readBytes;
import static org.junit.jupiter.api.Assertions.assertArrayEquals;
import static org.junit.jupiter.api.Assertions.assertEquals;
@@ -186,6 +190,76 @@ public class UtilsTest {
assertThrows(SerializationException.class, () -> readBytes(buf, 1));
}
+ @Test
+ public void shouldExtractHeadersTimestampPrefixWithHeaders() {
+ final byte[] headers = headersOf(HEADERS);
+ final byte[] input = headersTimestampValueOf(headers, VALUE);
+
+ // The prefix is everything rawPlainValue() strips:
[headersSize(varint)][headers][timestamp].
+ final byte[] expected = new byte[headers.length +
StateSerdes.TIMESTAMP_SIZE];
+ ByteBuffer.wrap(expected).put(headers).putLong(TIMESTAMP);
+
+ assertArrayEquals(expected, rawHeadersTimestampPrefix(input));
+ }
+
+ @Test
+ public void shouldExtractHeadersTimestampPrefixWithEmptyHeaders() {
+ // Empty headers encode as a single-byte varint, so the prefix is 1 +
8 bytes.
+ final byte[] prefix =
rawHeadersTimestampPrefix(timestampedValueWithEmptyHeaders(VALUE));
+
+ assertEquals(MIN_SIZE, prefix.length);
+ assertEquals((byte) 0x00, prefix[0]);
+ assertEquals(TIMESTAMP, ByteBuffer.wrap(prefix, 1,
StateSerdes.TIMESTAMP_SIZE).getLong());
+ }
+
+ @Test
+ public void shouldReturnNullForNullRawHeadersTimestampPrefix() {
+ assertNull(rawHeadersTimestampPrefix(null));
+ }
+
+ @Test
+ public void shouldRebuildValueTimestampHeadersFromPrefixAndPlainValue() {
+ final byte[] input = headersTimestampValueOf(headersOf(HEADERS),
VALUE);
+
+ // Splitting and rejoining must be lossless, for headers and for the
empty-headers case.
+ assertArrayEquals(input,
rawValueTimestampHeaders(rawHeadersTimestampPrefix(input),
rawPlainValue(input)));
+
+ final byte[] empty = timestampedValueWithEmptyHeaders(VALUE);
+ assertArrayEquals(empty,
rawValueTimestampHeaders(rawHeadersTimestampPrefix(empty),
rawPlainValue(empty)));
+ }
+
+ @Test
+ public void shouldReturnNullWhenRebuildingFromNullPlainValue() {
+
assertNull(rawValueTimestampHeaders(rawHeadersTimestampPrefix(timestampedValueWithEmptyHeaders(VALUE)),
null));
+ }
+
+ @Test
+ public void
shouldBuildValueTimestampHeadersFromPlainValueTimestampAndHeaders() {
+ final Headers headers = new RecordHeaders(new Header[]{new
RecordHeader("h", "hv".getBytes(StandardCharsets.UTF_8))});
+ final byte[] built = rawValueTimestampHeaders(VALUE, TIMESTAMP,
headers);
+
+ // Building must be the exact inverse of the extracting helpers.
+ assertArrayEquals(VALUE, rawPlainValue(built));
+ assertEquals(TIMESTAMP, Utils.timestamp(built));
+ assertEquals(headers, Utils.headers(built));
+ }
+
+ @Test
+ public void
shouldBuildValueTimestampHeadersWithEmptyHeadersLikeTheTimestampOnlyOverload() {
+ // Empty and null headers must both encode as headersSize=0, i.e.
exactly what the overload
+ // without headers produces -- otherwise the two writers would
disagree on the wire format.
+ final byte[] expected = timestampedValueWithEmptyHeaders(VALUE);
+
+ assertArrayEquals(expected, rawValueTimestampHeaders(VALUE,
TIMESTAMP));
+ assertArrayEquals(expected, rawValueTimestampHeaders(VALUE, TIMESTAMP,
new RecordHeaders()));
+ assertArrayEquals(expected, rawValueTimestampHeaders(VALUE, TIMESTAMP,
null));
+ }
+
+ @Test
+ public void shouldReturnNullWhenBuildingFromNullPlainValueWithHeaders() {
+ assertNull(rawValueTimestampHeaders(null, TIMESTAMP, new
RecordHeaders()));
+ }
+
private static byte[] timestampedValueWithEmptyHeaders(final byte[] value)
{
// header size: 1 byte, empty headers: 0 byte, timestamp: 8 bytes,
plain value length
final byte[] res = new byte[1 + 0 + StateSerdes.TIMESTAMP_SIZE +
value.length];
