You are viewing a plain text version of this content. The canonical link for it is here.
Posted to jira@kafka.apache.org by "aliehsaeedii (via GitHub)" <gi...@apache.org> on 2026/04/28 09:44:49 UTC
[PR] KAFKA-20413:: Make Streams suppress headers-aware [kafka]
aliehsaeedii opened a new pull request, #22165:
URL: https://github.com/apache/kafka/pull/22165
This PR fixes the suppress-buffer part of [KAFKA-20413](https://issues.apache.org/jira/browse/KAFKA-20413): `InMemoryTimeOrderedKeyValueChangeBuffer` now uses the buffered record's own headers in serde calls and on eviction, and `priorValueForBuffered` returns the real timestamp + headers instead of `RecordQueue.UNKNOWN` / empty `RecordHeaders`.
Testing:
Added `shouldPropagateHeadersThroughEviction`, `shouldReturnBufferedHeadersFromPriorValue`, `shouldUseRecordHeadersNotProcessorContextHeadersOnPut`.
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-20413: Make Streams suppress headers-aware [kafka]
Posted by "aliehsaeedii (via GitHub)" <gi...@apache.org>.
aliehsaeedii commented on code in PR #22165:
URL: https://github.com/apache/kafka/pull/22165#discussion_r3656415254
##########
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:
Good catch, fixed.
One note: the tombstone case turns out to be order-independent, the newValue == null returns early without touching the headers, so only the old value writes either way. The order matters for the ordinary update path.
There is a tombstone question in headers mode, though, and I'd like your view: there the old value is serialized against the headers recovered from the previous new value, not against the live record headers, so a tombstone leaves the emitted headers unwritten where plain mode would carry the old value's. On eviction I fall back to the buffered context's headers. Is that the behaviour we'd want, or should headers mode match plain mode here?
##########
streams/src/main/java/org/apache/kafka/streams/state/internals/InMemoryTimeOrderedKeyValueChangeBuffer.java:
##########
@@ -71,8 +75,14 @@ public final class InMemoryTimeOrderedKeyValueChangeBuffer<K, V, T> implements T
private static final byte[] V_1_CHANGELOG_HEADER_VALUE = {(byte) 1};
private static final byte[] V_2_CHANGELOG_HEADER_VALUE = {(byte) 2};
private static final byte[] V_3_CHANGELOG_HEADER_VALUE = {(byte) 3};
+ // V4 is identical to V3 on the wire, except that the prior/old/new value parts are each encoded
+ // as a ValueTimestampHeaders blob ([headersSize][headers][timestamp][value]) instead of a plain
+ // value. It is only written when headers-aware stores are enabled (dsl.store.format=HEADERS).
+ private static final byte[] V_4_CHANGELOG_HEADER_VALUE = {(byte) 4};
Review Comment:
you're right!
But a suppress entry holds three value parts (prior/old/new) that can each come from a different record, so it's not a single header set like the other stores — I'll use distinct record-header keys per part. We'd use distinct keys (vh.old, vh.prior, …) rather than one.
##########
streams/src/main/java/org/apache/kafka/streams/state/internals/InMemoryTimeOrderedKeyValueChangeBuffer.java:
##########
@@ -465,28 +513,165 @@ 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 ? unwrapHeadersFormatToPlainValue(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(
+ wrapPlainValueAsHeadersFormat(value.priorValue(), timestamp),
+ wrapPlainValueAsHeadersFormat(value.oldValue(), timestamp),
+ wrapPlainValueAsHeadersFormat(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);
+ }
+
+ // Wraps a plain value part as a ValueTimestampHeaders blob ([headersSize=0][timestamp][value])
+ // without needing a value serde. Used to normalize restored V0-V3 records to the in-memory
+ // encoding used when header stores are enabled.
+ private static byte[] wrapPlainValueAsHeadersFormat(final byte[] plainValue, final long timestamp) {
+ if (plainValue == null) {
+ return null;
+ }
+ final ByteBuffer buffer = ByteBuffer.allocate(ByteUtils.sizeOfVarint(0) + Long.BYTES + plainValue.length);
+ ByteUtils.writeVarint(0, buffer); // empty headers
+ buffer.putLong(timestamp);
+ buffer.put(plainValue);
+ return buffer.array();
+ }
+
+ // Strips the ValueTimestampHeaders wrapper ([headersSize][headers][timestamp]) from a value
+ // part, leaving the plain value bytes. Used to normalize restored V4 records to the in-memory
+ // encoding used when header stores are disabled.
+ private static byte[] unwrapHeadersFormatToPlainValue(final byte[] headersFormatValue) {
+ if (headersFormatValue == null) {
+ return null;
+ }
+ final ByteBuffer buffer = ByteBuffer.wrap(headersFormatValue);
+ final int headersSize = ByteUtils.readVarint(buffer);
+ buffer.position(buffer.position() + headersSize + Long.BYTES);
+ final byte[] plainValue = new byte[buffer.remaining()];
+ buffer.get(plainValue);
+ return plainValue;
+ }
+
@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 newHeaders = new RecordHeaders(record.headers());
+ final long newTimestamp = record.timestamp();
+ final Bytes serializedKey = Bytes.wrap(keySerde.serializer().serialize(changelogTopic, newHeaders, 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(
Review Comment:
You're right — reverted. The recordContext.headers() already is record.headers(); effectiveContext was a no-op that additionally dropped sourceRawKey/sourceRawValue via the 5-arg constructor. On evict I now read each part's ValueTimestampHeaders blob directly, which is what actually matters for the old value (it comes from an earlier record than the new one).
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-20413: Make Streams suppress headers-aware [kafka]
Posted by "mjsax (via GitHub)" <gi...@apache.org>.
mjsax commented on code in PR #22165:
URL: https://github.com/apache/kafka/pull/22165#discussion_r3560337815
##########
streams/src/main/java/org/apache/kafka/streams/state/internals/InMemoryTimeOrderedKeyValueChangeBuffer.java:
##########
@@ -437,19 +436,22 @@ public void evictWhile(final Supplier<Boolean> predicate,
@Override
public Maybe<ValueTimestampHeaders<V>> priorValueForBuffered(final K key) {
final Bytes serializedKey = Bytes.wrap(keySerde.serializer().serialize(changelogTopic, context.headers(), key));
Review Comment:
It's the same question we had on other PRs, -- we decided to just use `context.headers()`.
I think, empty headers might also be valid, but given that we decided to use `context.headers()` elsewhere, might be best to keep as-is for consistency.
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-20413: Make Streams suppress headers-aware [kafka]
Posted by "mjsax (via GitHub)" <gi...@apache.org>.
mjsax merged PR #22165:
URL: https://github.com/apache/kafka/pull/22165
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-20413: Make Streams suppress headers-aware [kafka]
Posted by "aliehsaeedii (via GitHub)" <gi...@apache.org>.
aliehsaeedii commented on code in PR #22165:
URL: https://github.com/apache/kafka/pull/22165#discussion_r3638367418
##########
streams/src/main/java/org/apache/kafka/streams/state/internals/InMemoryTimeOrderedKeyValueChangeBuffer.java:
##########
@@ -399,12 +438,12 @@ 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())
+ final Headers headers = bufferValue.context().headers();
Review Comment:
in headers mode each part is already deserialized as a ValueTimestampHeaders and uses its own embedded headers; the headers var only feeds the key deserialize and the plain-mode value path. The `bufferValue.newValue().headers()` won't compile I assume.
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-20413: Make Streams suppress headers-aware [kafka]
Posted by "mjsax (via GitHub)" <gi...@apache.org>.
mjsax commented on code in PR #22165:
URL: https://github.com/apache/kafka/pull/22165#discussion_r3654754480
##########
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:
We actually need to serialize oldValue first (same as `FullChangeSerde#serializeParts` which is called in the old code further above, to ensure that tombstone are emitted with the correct header).
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-20413: Make Streams suppress headers-aware [kafka]
Posted by "aliehsaeedii (via GitHub)" <gi...@apache.org>.
aliehsaeedii commented on code in PR #22165:
URL: https://github.com/apache/kafka/pull/22165#discussion_r3558729010
##########
streams/src/main/java/org/apache/kafka/streams/state/internals/InMemoryTimeOrderedKeyValueChangeBuffer.java:
##########
@@ -437,19 +436,22 @@ public void evictWhile(final Supplier<Boolean> predicate,
@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();
final V deserializedValue = valueSerde.innerSerde().deserializer().deserialize(
changelogTopic,
- context.headers(),
+ bufferValue.context().headers(),
serializedValue
);
- // it's unfortunately not possible to know this, unless we materialize the suppressed result, since our only
- // knowledge of the prior value is what the upstream processor sends us as the "old value" when we first
- // buffer something.
- return Maybe.defined(ValueTimestampHeaders.make(deserializedValue, RecordQueue.UNKNOWN, new RecordHeaders()));
+ return Maybe.defined(ValueTimestampHeaders.make(
+ deserializedValue,
+ bufferValue.context().timestamp(),
Review Comment:
Due to these explanations I reverted the changes of here. The former [commit](https://github.com/apache/kafka/pull/22165/changes/87bee53184f012260b5bad0c293e653a3601c7e9#diff-b94df16c1e8d51db85b6539dce34da2e6cfda0a42584607d60c867e543f268e7) returns `bufferValue.context().timestamp()` instead of `RecordQueue.UNKNOWN` and `RecordHeaders(bufferValue.context().headers())` instead of empty headers.
##########
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()))));
Review Comment:
These only cover a single put per key. The value-vs-metadata mismatch shows up when the same key is put twice before eviction (prior value from put #1, timestamp/headers from put #2) — worth a test for that case.
##########
streams/src/main/java/org/apache/kafka/streams/state/internals/InMemoryTimeOrderedKeyValueChangeBuffer.java:
##########
@@ -437,19 +436,22 @@ public void evictWhile(final Supplier<Boolean> predicate,
@Override
public Maybe<ValueTimestampHeaders<V>> priorValueForBuffered(final K key) {
final Bytes serializedKey = Bytes.wrap(keySerde.serializer().serialize(changelogTopic, context.headers(), key));
Review Comment:
I'm wondering if `context.headers()` is the right thing for serializing the key here and if not, how to replace it with the correct one?! Maybe just sending empty headers?
##########
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:
Every getBytes(...) call in this file specifies UTF_8 and UTF_8 is already imported. Keeping getBytes(UTF_8) matches that convention and avoids relying on the platform default charset.
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-20413: Make Streams suppress headers-aware [kafka]
Posted by "aliehsaeedii (via GitHub)" <gi...@apache.org>.
aliehsaeedii commented on code in PR #22165:
URL: https://github.com/apache/kafka/pull/22165#discussion_r3559281841
##########
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:
All other calls are using `utf8` too, so I keep it as-is:)
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-20413: Make Streams suppress headers-aware [kafka]
Posted by "mjsax (via GitHub)" <gi...@apache.org>.
mjsax commented on code in PR #22165:
URL: https://github.com/apache/kafka/pull/22165#discussion_r3591119485
##########
streams/src/test/java/org/apache/kafka/streams/kstream/internals/SuppressScenarioTest.java:
##########
@@ -161,6 +164,45 @@ public void shouldImmediatelyEmitEventsWithZeroEmitAfter() {
}
}
+ @Test
+ public void shouldPropagateHeadersThroughSuppression() {
+ final StreamsBuilder builder = new StreamsBuilder();
+
+ final KTable<String, Long> valueCounts = builder
+ .table(
+ "input",
+ Consumed.with(STRING_SERDE, STRING_SERDE),
+ Materialized.<String, String, KeyValueStore<Bytes, byte[]>>with(STRING_SERDE, STRING_SERDE)
+ .withCachingDisabled()
+ .withLoggingDisabled()
+ )
+ .groupBy((k, v) -> new KeyValue<>(v, k), Grouped.with(STRING_SERDE, STRING_SERDE))
+ .count();
+
+ valueCounts
+ .suppress(untilTimeLimit(ZERO, unbounded()))
Review Comment:
I don't think this work correctly. If we set `ZERO` we flush the suppress buffer right away, and even in the old code, headers would get carried forward via the context... (I just ran this this unmodified on `trunk` and it also passed).
Instead, we need to set an actual suppress time and pipe two input record. The second record (with larger ts) would flush the suppress buffer emitting the first. W/o the fix, the headers from the second record would carry forward into the result, but with the fix, the correct headers from the suppressed record should be there.
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-20413: Make Streams suppress headers-aware [kafka]
Posted by "aliehsaeedii (via GitHub)" <gi...@apache.org>.
aliehsaeedii commented on code in PR #22165:
URL: https://github.com/apache/kafka/pull/22165#discussion_r3638547075
##########
streams/src/main/java/org/apache/kafka/streams/state/internals/InMemoryTimeOrderedKeyValueChangeBuffer.java:
##########
@@ -465,28 +513,165 @@ 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 ? unwrapHeadersFormatToPlainValue(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(
+ wrapPlainValueAsHeadersFormat(value.priorValue(), timestamp),
+ wrapPlainValueAsHeadersFormat(value.oldValue(), timestamp),
+ wrapPlainValueAsHeadersFormat(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);
+ }
+
+ // Wraps a plain value part as a ValueTimestampHeaders blob ([headersSize=0][timestamp][value])
+ // without needing a value serde. Used to normalize restored V0-V3 records to the in-memory
+ // encoding used when header stores are enabled.
+ private static byte[] wrapPlainValueAsHeadersFormat(final byte[] plainValue, final long timestamp) {
+ if (plainValue == null) {
+ return null;
+ }
+ final ByteBuffer buffer = ByteBuffer.allocate(ByteUtils.sizeOfVarint(0) + Long.BYTES + plainValue.length);
+ ByteUtils.writeVarint(0, buffer); // empty headers
+ buffer.putLong(timestamp);
+ buffer.put(plainValue);
+ return buffer.array();
+ }
+
+ // Strips the ValueTimestampHeaders wrapper ([headersSize][headers][timestamp]) from a value
+ // part, leaving the plain value bytes. Used to normalize restored V4 records to the in-memory
+ // encoding used when header stores are disabled.
+ private static byte[] unwrapHeadersFormatToPlainValue(final byte[] headersFormatValue) {
+ if (headersFormatValue == null) {
+ return null;
+ }
+ final ByteBuffer buffer = ByteBuffer.wrap(headersFormatValue);
+ final int headersSize = ByteUtils.readVarint(buffer);
+ buffer.position(buffer.position() + headersSize + Long.BYTES);
+ final byte[] plainValue = new byte[buffer.remaining()];
+ buffer.get(plainValue);
+ return plainValue;
+ }
+
@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 newHeaders = new RecordHeaders(record.headers());
+ final long newTimestamp = record.timestamp();
+ final Bytes serializedKey = Bytes.wrap(keySerde.serializer().serialize(changelogTopic, newHeaders, 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(),
+ newHeaders
+ );
+
+ // 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.
Review Comment:
I agree for `oldValue == null` case, but the block still matters for a first insert where the key pre-existed.
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-20413: Make Streams suppress headers-aware [kafka]
Posted by "mjsax (via GitHub)" <gi...@apache.org>.
mjsax commented on code in PR #22165:
URL: https://github.com/apache/kafka/pull/22165#discussion_r3591889688
##########
streams/src/test/java/org/apache/kafka/streams/kstream/internals/SuppressScenarioTest.java:
##########
@@ -161,6 +164,45 @@ public void shouldImmediatelyEmitEventsWithZeroEmitAfter() {
}
}
+ @Test
+ public void shouldPropagateHeadersThroughSuppression() {
+ final StreamsBuilder builder = new StreamsBuilder();
+
+ final KTable<String, Long> valueCounts = builder
+ .table(
+ "input",
+ Consumed.with(STRING_SERDE, STRING_SERDE),
+ Materialized.<String, String, KeyValueStore<Bytes, byte[]>>with(STRING_SERDE, STRING_SERDE)
+ .withCachingDisabled()
+ .withLoggingDisabled()
+ )
+ .groupBy((k, v) -> new KeyValue<>(v, k), Grouped.with(STRING_SERDE, STRING_SERDE))
+ .count();
+
+ valueCounts
+ .suppress(untilTimeLimit(ZERO, unbounded()))
+ .toStream()
+ .to("output-suppressed", Produced.with(STRING_SERDE, Serdes.Long()));
Review Comment:
I am also wondering, if we should add another `process()` step before we write to the output topic, to access `context.headers()` (and maybe add information about it to the record's value, so we can capture it on the output topic).
When the suppress() flush happens while processing the second record, the context should not have been set to the flushed/first record's header I believe, but should be the headers from the second record triggering the flush? Not 100% sure.
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-20413: Make Streams suppress headers-aware [kafka]
Posted by "aliehsaeedii (via GitHub)" <gi...@apache.org>.
aliehsaeedii commented on code in PR #22165:
URL: https://github.com/apache/kafka/pull/22165#discussion_r3638440066
##########
streams/src/main/java/org/apache/kafka/streams/state/internals/InMemoryTimeOrderedKeyValueChangeBuffer.java:
##########
@@ -465,28 +513,165 @@ 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 ? unwrapHeadersFormatToPlainValue(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(
+ wrapPlainValueAsHeadersFormat(value.priorValue(), timestamp),
+ wrapPlainValueAsHeadersFormat(value.oldValue(), timestamp),
+ wrapPlainValueAsHeadersFormat(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);
+ }
+
+ // Wraps a plain value part as a ValueTimestampHeaders blob ([headersSize=0][timestamp][value])
+ // without needing a value serde. Used to normalize restored V0-V3 records to the in-memory
+ // encoding used when header stores are enabled.
+ private static byte[] wrapPlainValueAsHeadersFormat(final byte[] plainValue, final long timestamp) {
+ if (plainValue == null) {
+ return null;
+ }
+ final ByteBuffer buffer = ByteBuffer.allocate(ByteUtils.sizeOfVarint(0) + Long.BYTES + plainValue.length);
+ ByteUtils.writeVarint(0, buffer); // empty headers
+ buffer.putLong(timestamp);
+ buffer.put(plainValue);
+ return buffer.array();
+ }
+
+ // Strips the ValueTimestampHeaders wrapper ([headersSize][headers][timestamp]) from a value
+ // part, leaving the plain value bytes. Used to normalize restored V4 records to the in-memory
+ // encoding used when header stores are disabled.
+ private static byte[] unwrapHeadersFormatToPlainValue(final byte[] headersFormatValue) {
+ if (headersFormatValue == null) {
+ return null;
+ }
+ final ByteBuffer buffer = ByteBuffer.wrap(headersFormatValue);
+ final int headersSize = ByteUtils.readVarint(buffer);
+ buffer.position(buffer.position() + headersSize + Long.BYTES);
+ final byte[] plainValue = new byte[buffer.remaining()];
+ buffer.get(plainValue);
+ return plainValue;
+ }
+
@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 newHeaders = new RecordHeaders(record.headers());
+ final long newTimestamp = record.timestamp();
+ final Bytes serializedKey = Bytes.wrap(keySerde.serializer().serialize(changelogTopic, newHeaders, 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(
Review Comment:
Plain-mode still needs the context?!
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-20413: Make Streams suppress headers-aware [kafka]
Posted by "aliehsaeedii (via GitHub)" <gi...@apache.org>.
aliehsaeedii commented on code in PR #22165:
URL: https://github.com/apache/kafka/pull/22165#discussion_r3638805538
##########
streams/src/main/java/org/apache/kafka/streams/state/internals/InMemoryTimeOrderedKeyValueChangeBuffer.java:
##########
@@ -465,28 +513,165 @@ 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 ? unwrapHeadersFormatToPlainValue(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(
+ wrapPlainValueAsHeadersFormat(value.priorValue(), timestamp),
+ wrapPlainValueAsHeadersFormat(value.oldValue(), timestamp),
+ wrapPlainValueAsHeadersFormat(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);
+ }
+
+ // Wraps a plain value part as a ValueTimestampHeaders blob ([headersSize=0][timestamp][value])
+ // without needing a value serde. Used to normalize restored V0-V3 records to the in-memory
+ // encoding used when header stores are enabled.
+ private static byte[] wrapPlainValueAsHeadersFormat(final byte[] plainValue, final long timestamp) {
+ if (plainValue == null) {
+ return null;
+ }
+ final ByteBuffer buffer = ByteBuffer.allocate(ByteUtils.sizeOfVarint(0) + Long.BYTES + plainValue.length);
+ ByteUtils.writeVarint(0, buffer); // empty headers
+ buffer.putLong(timestamp);
+ buffer.put(plainValue);
+ return buffer.array();
+ }
+
+ // Strips the ValueTimestampHeaders wrapper ([headersSize][headers][timestamp]) from a value
+ // part, leaving the plain value bytes. Used to normalize restored V4 records to the in-memory
+ // encoding used when header stores are disabled.
+ private static byte[] unwrapHeadersFormatToPlainValue(final byte[] headersFormatValue) {
+ if (headersFormatValue == null) {
+ return null;
+ }
+ final ByteBuffer buffer = ByteBuffer.wrap(headersFormatValue);
+ final int headersSize = ByteUtils.readVarint(buffer);
+ buffer.position(buffer.position() + headersSize + Long.BYTES);
+ final byte[] plainValue = new byte[buffer.remaining()];
+ buffer.get(plainValue);
+ return plainValue;
+ }
+
@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 newHeaders = new RecordHeaders(record.headers());
+ final long newTimestamp = record.timestamp();
+ final Bytes serializedKey = Bytes.wrap(keySerde.serializer().serialize(changelogTopic, newHeaders, 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(),
+ newHeaders
+ );
+
+ // 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.
Review Comment:
See https://github.com/apache/kafka/pull/22165/changes#r3638697901
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-20413: Make Streams suppress headers-aware [kafka]
Posted by "mjsax (via GitHub)" <gi...@apache.org>.
mjsax commented on code in PR #22165:
URL: https://github.com/apache/kafka/pull/22165#discussion_r3653777426
##########
streams/src/main/java/org/apache/kafka/streams/state/internals/InMemoryTimeOrderedKeyValueChangeBuffer.java:
##########
@@ -399,12 +438,12 @@ 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())
+ final Headers headers = bufferValue.context().headers();
Review Comment:
> Otherwise we don't pass the correct record header into this deserialization step.
Maybe I am wrong here, and `bufferValue.context().headers()` is actually the same as the header extracted from `bufferValue.newValue()`, so we are good here.
Might be good to add a comment about it.
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-20413: Make Streams suppress headers-aware [kafka]
Posted by "mjsax (via GitHub)" <gi...@apache.org>.
mjsax commented on code in PR #22165:
URL: https://github.com/apache/kafka/pull/22165#discussion_r3653777426
##########
streams/src/main/java/org/apache/kafka/streams/state/internals/InMemoryTimeOrderedKeyValueChangeBuffer.java:
##########
@@ -399,12 +438,12 @@ 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())
+ final Headers headers = bufferValue.context().headers();
Review Comment:
> Otherwise we don't pass the correct record header into this deserialization step.
Maybe I am wrong here, and `bufferValue.context().headers()` is actually the same as the header extracted from `bufferValue.newValue()`, so we are good here.
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-20413: Make Streams suppress headers-aware [kafka]
Posted by "mjsax (via GitHub)" <gi...@apache.org>.
mjsax commented on code in PR #22165:
URL: https://github.com/apache/kafka/pull/22165#discussion_r3654675708
##########
streams/src/main/java/org/apache/kafka/streams/state/internals/InMemoryTimeOrderedKeyValueChangeBuffer.java:
##########
@@ -71,8 +75,14 @@ public final class InMemoryTimeOrderedKeyValueChangeBuffer<K, V, T> implements T
private static final byte[] V_1_CHANGELOG_HEADER_VALUE = {(byte) 1};
private static final byte[] V_2_CHANGELOG_HEADER_VALUE = {(byte) 2};
private static final byte[] V_3_CHANGELOG_HEADER_VALUE = {(byte) 3};
+ // V4 is identical to V3 on the wire, except that the prior/old/new value parts are each encoded
+ // as a ValueTimestampHeaders blob ([headersSize][headers][timestamp][value]) instead of a plain
+ // value. It is only written when headers-aware stores are enabled (dsl.store.format=HEADERS).
+ private static final byte[] V_4_CHANGELOG_HEADER_VALUE = {(byte) 4};
Review Comment:
Thinking about it, introducing a V4 would break our offline downgrade path, right? KS 4.3 and older won't be able to read V4 format.
It seems we should not add V4, but put the header into the Kafka message header field, as we did for all other state store to keep the value-bytes unmodified, and backwards compatible.
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-20413: Make Streams suppress headers-aware [kafka]
Posted by "UladzislauBlok (via GitHub)" <gi...@apache.org>.
UladzislauBlok commented on code in PR #22165:
URL: https://github.com/apache/kafka/pull/22165#discussion_r3163929068
##########
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:
minor: we don't need to specify utf8
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-20413: Make Streams suppress headers-aware [kafka]
Posted by "mjsax (via GitHub)" <gi...@apache.org>.
mjsax commented on code in PR #22165:
URL: https://github.com/apache/kafka/pull/22165#discussion_r3626759807
##########
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 it sufficient/correct to check the config? Should we rather consult the upstream `KTable` store if it has headers enabled or not, and align the suppress buffer accordingly? A user might not set the config, but enable headers via `Materialized` upstream?
We might also want to to a follow up KIP, to extend `suppress()` by itself to have an operator based on/off switch? Of course, out-of-scope for this PR.
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-20413: Make Streams suppress headers-aware [kafka]
Posted by "aliehsaeedii (via GitHub)" <gi...@apache.org>.
aliehsaeedii commented on code in PR #22165:
URL: https://github.com/apache/kafka/pull/22165#discussion_r3558729017
##########
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()))));
Review Comment:
These only cover a single put per key. The value-vs-metadata mismatch shows up when the same key is put twice before eviction (prior value from put #1, timestamp/headers from put #2) — worth a test for that case.
##########
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:
Every getBytes(...) call in this file specifies UTF_8 and UTF_8 is already imported. Keeping getBytes(UTF_8) matches that convention and avoids relying on the platform default charset.
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-20413: Make Streams suppress headers-aware [kafka]
Posted by "mjsax (via GitHub)" <gi...@apache.org>.
mjsax commented on code in PR #22165:
URL: https://github.com/apache/kafka/pull/22165#discussion_r3606501115
##########
streams/src/main/java/org/apache/kafka/streams/state/internals/InMemoryTimeOrderedKeyValueChangeBuffer.java:
##########
@@ -472,21 +474,31 @@ public boolean put(final long time,
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());
-
+ final RecordHeaders incomingHeaders = new RecordHeaders(record.headers());
+ final Bytes serializedKey = Bytes.wrap(keySerde.serializer().serialize(changelogTopic, incomingHeaders, record.key()));
final BufferValue buffered = getBuffered(serializedKey);
- final byte[] serializedPriorValue;
- if (buffered == null) {
- serializedPriorValue = serialChange.oldValue;
- } else {
- serializedPriorValue = buffered.priorValue();
- }
+
+ // The headers emitted on eviction must correspond to the record's key, which is anchored to
+ // the record that first created this buffer slot (the same record that anchors the buffer-time
+ // and the prior value). So on an in-place update we keep the original entry's headers rather
+ // than adopting the newer record's headers
+ final RecordHeaders bufferedHeaders =
+ buffered == null ? incomingHeaders : new RecordHeaders(buffered.context().headers());
+ final ProcessorRecordContext effectiveContext = new ProcessorRecordContext(
+ recordContext.timestamp(),
+ recordContext.offset(),
+ recordContext.partition(),
+ recordContext.topic(),
+ bufferedHeaders
+ );
+
+ final Change<byte[]> serialChange = valueSerde.serializeParts(changelogTopic, bufferedHeaders, record.value());
+ final byte[] serializedPriorValue = buffered == null ? serialChange.oldValue : buffered.priorValue();
cleanPut(
time,
serializedKey,
- new BufferValue(serializedPriorValue, serialChange.oldValue, serialChange.newValue, recordContext)
+ new BufferValue(serializedPriorValue, serialChange.oldValue, serialChange.newValue, effectiveContext)
Review Comment:
I am confuse the this change -- when we do a `put()` we need to put the headers for the new record, not the old record, right?
I did more digging and it seems we need a different fix... What the buffer stores is an "oldValue" and a "newValue" but only one record-context for both. Your change is changing what we put into the record context, but I believe that is not what we should change -- the context should be from the currently processes record.
Thus, we actually need to change the serialization format of `oldValue` and `newValue` -- atm, we serializer plain value types, but we would need to start using `ValueTimestampHeader` type (this would also fix another issue that we atm also lose the correct record timestamp via suppress). This allows us to store both header for old and new and get the correct old headers into the output on eviction.
So we need to update `keySerde` and `valueSerde` and create a corresponding `ValueTimestampHeaders` object here that we put into the suppress buffer.
For backward compatibility, we should also do this only if header stores are enable I believe? Otherwise, we would increase users' in-memory consumption significantly.
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-20413: Make Streams suppress headers-aware [kafka]
Posted by "mjsax (via GitHub)" <gi...@apache.org>.
mjsax commented on code in PR #22165:
URL: https://github.com/apache/kafka/pull/22165#discussion_r3591889688
##########
streams/src/test/java/org/apache/kafka/streams/kstream/internals/SuppressScenarioTest.java:
##########
@@ -161,6 +164,45 @@ public void shouldImmediatelyEmitEventsWithZeroEmitAfter() {
}
}
+ @Test
+ public void shouldPropagateHeadersThroughSuppression() {
+ final StreamsBuilder builder = new StreamsBuilder();
+
+ final KTable<String, Long> valueCounts = builder
+ .table(
+ "input",
+ Consumed.with(STRING_SERDE, STRING_SERDE),
+ Materialized.<String, String, KeyValueStore<Bytes, byte[]>>with(STRING_SERDE, STRING_SERDE)
+ .withCachingDisabled()
+ .withLoggingDisabled()
+ )
+ .groupBy((k, v) -> new KeyValue<>(v, k), Grouped.with(STRING_SERDE, STRING_SERDE))
+ .count();
+
+ valueCounts
+ .suppress(untilTimeLimit(ZERO, unbounded()))
+ .toStream()
+ .to("output-suppressed", Produced.with(STRING_SERDE, Serdes.Long()));
Review Comment:
I am also wondering, if we should add another `process()` step before we write to the output topic, to access `context.headers()` (and maybe add information about it to the record's value, so we can capture it on the output topic).
When the suppress() flush happens while processing the second record, the context should have also be restored to the first record's header, right? Or would this check be redundant?
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-20413: Make Streams suppress headers-aware [kafka]
Posted by "aliehsaeedii (via GitHub)" <gi...@apache.org>.
aliehsaeedii commented on code in PR #22165:
URL: https://github.com/apache/kafka/pull/22165#discussion_r3595634626
##########
streams/src/test/java/org/apache/kafka/streams/kstream/internals/SuppressScenarioTest.java:
##########
@@ -161,6 +164,45 @@ public void shouldImmediatelyEmitEventsWithZeroEmitAfter() {
}
}
+ @Test
+ public void shouldPropagateHeadersThroughSuppression() {
+ final StreamsBuilder builder = new StreamsBuilder();
+
+ final KTable<String, Long> valueCounts = builder
+ .table(
+ "input",
+ Consumed.with(STRING_SERDE, STRING_SERDE),
+ Materialized.<String, String, KeyValueStore<Bytes, byte[]>>with(STRING_SERDE, STRING_SERDE)
+ .withCachingDisabled()
+ .withLoggingDisabled()
+ )
+ .groupBy((k, v) -> new KeyValue<>(v, k), Grouped.with(STRING_SERDE, STRING_SERDE))
+ .count();
+
+ valueCounts
+ .suppress(untilTimeLimit(ZERO, unbounded()))
Review Comment:
I verified the reworked scenario (2-record) and it still passes on trunk. The question is whether TTD is propagating/mutating the specific ProcessorRecordContext instance that gets stored inside k1's BufferValue when k1 (1st input record) is buffered? I assume not. I think I must remove the e2e test since it does not prove anything.
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-20413: Make Streams suppress headers-aware [kafka]
Posted by "aliehsaeedii (via GitHub)" <gi...@apache.org>.
aliehsaeedii commented on PR #22165:
URL: https://github.com/apache/kafka/pull/22165#issuecomment-4968398928
Build failed because of unrelated flaky tests!
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-20413: Make Streams suppress headers-aware [kafka]
Posted by "mjsax (via GitHub)" <gi...@apache.org>.
mjsax commented on code in PR #22165:
URL: https://github.com/apache/kafka/pull/22165#discussion_r3671439944
##########
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:
I filed a ticket for it: https://issues.apache.org/jira/browse/KAFKA-20856 -- yes, the idea is to overwrite the config, and yes it would need a KIP. It's not anything critical; just filed for completeness. Don't think there is high pressure to pick it up right away.
We can do the same via `Materialized` by passing in a non-header supplier -- even if the DSL configs says headers store (which would pick a headers-supplier), a custom supplier would take preference effectively disabling (or enabling) headers and the config does not take effect.
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-20413: Make Streams suppress headers-aware [kafka]
Posted by "aliehsaeedii (via GitHub)" <gi...@apache.org>.
aliehsaeedii commented on code in PR #22165:
URL: https://github.com/apache/kafka/pull/22165#discussion_r3638790342
##########
streams/src/main/java/org/apache/kafka/streams/state/internals/InMemoryTimeOrderedKeyValueChangeBuffer.java:
##########
@@ -437,12 +476,21 @@ public void evictWhile(final Supplier<Boolean> predicate,
@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).
Review Comment:
the first time, the old value's ValueTimestampHeaders is collapsed to .value() in the upstream KTable processor. So the main issue is [here](https://github.com/apache/kafka/blob/5a018810d42742615b83155a60086047c6611dfd/streams/src/main/java/org/apache/kafka/streams/kstream/internals/KTableSource.java#L145). I'm wondering why we did not use the headers in this class at all even though we chnaged ValueAndTmestamp to VTH
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-20413: Make Streams suppress headers-aware [kafka]
Posted by "mjsax (via GitHub)" <gi...@apache.org>.
mjsax commented on code in PR #22165:
URL: https://github.com/apache/kafka/pull/22165#discussion_r3653576654
##########
streams/src/main/java/org/apache/kafka/streams/state/internals/InMemoryTimeOrderedKeyValueChangeBuffer.java:
##########
@@ -399,12 +438,12 @@ 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())
+ final Headers headers = bufferValue.context().headers();
Review Comment:
> in headers mode each part is already deserialized as a ValueTimestampHeaders and uses its own embedded headers
Yes, it would require to extract the header again.
> the headers var only feeds the key deserialize
Yes, but this seems to be the critical part, that must work correctly? Otherwise we don't pass the correct record header into this deserialization step.
> The bufferValue.newValue().headers() won't compile I assume.
There is no such method -- `newValue()` only returns `byte[]`, but we have existing helpers that can extract the headers from the byte[] array.
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-20413: Make Streams suppress headers-aware [kafka]
Posted by "mjsax (via GitHub)" <gi...@apache.org>.
mjsax commented on code in PR #22165:
URL: https://github.com/apache/kafka/pull/22165#discussion_r3654652028
##########
streams/src/main/java/org/apache/kafka/streams/state/internals/InMemoryTimeOrderedKeyValueChangeBuffer.java:
##########
@@ -399,12 +438,12 @@ 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())
+ final Headers headers = bufferValue.context().headers();
+ final K key = keySerde.deserializer().deserialize(changelogTopic, headers, next.getKey().key().get());
+ final Change<V> value = new Change<>(
+ deserializeValue(bufferValue.newValue(), headers),
+ deserializeValue(bufferValue.oldValue(), headers)
Review Comment:
Ah. Yes. My bad. It's a single record, and we don't really have a header for `oldValue`.
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-20413: Make Streams suppress headers-aware [kafka]
Posted by "mjsax (via GitHub)" <gi...@apache.org>.
mjsax commented on code in PR #22165:
URL: https://github.com/apache/kafka/pull/22165#discussion_r3654664458
##########
streams/src/main/java/org/apache/kafka/streams/state/internals/InMemoryTimeOrderedKeyValueChangeBuffer.java:
##########
@@ -437,12 +476,21 @@ public void evictWhile(final Supplier<Boolean> predicate,
@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).
Review Comment:
I guess we just inherit very old KS behavior... A `Change<V>` is only design for a plain value, and new/old have naturally the same type. -- This code was written with 0.10.0.0 release 10 years ago...
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-20413: Make Streams suppress headers-aware [kafka]
Posted by "mjsax (via GitHub)" <gi...@apache.org>.
mjsax commented on code in PR #22165:
URL: https://github.com/apache/kafka/pull/22165#discussion_r3654695466
##########
streams/src/main/java/org/apache/kafka/streams/state/internals/InMemoryTimeOrderedKeyValueChangeBuffer.java:
##########
@@ -465,28 +513,165 @@ 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 ? unwrapHeadersFormatToPlainValue(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(
+ wrapPlainValueAsHeadersFormat(value.priorValue(), timestamp),
+ wrapPlainValueAsHeadersFormat(value.oldValue(), timestamp),
+ wrapPlainValueAsHeadersFormat(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);
+ }
+
+ // Wraps a plain value part as a ValueTimestampHeaders blob ([headersSize=0][timestamp][value])
+ // without needing a value serde. Used to normalize restored V0-V3 records to the in-memory
+ // encoding used when header stores are enabled.
+ private static byte[] wrapPlainValueAsHeadersFormat(final byte[] plainValue, final long timestamp) {
+ if (plainValue == null) {
+ return null;
+ }
+ final ByteBuffer buffer = ByteBuffer.allocate(ByteUtils.sizeOfVarint(0) + Long.BYTES + plainValue.length);
+ ByteUtils.writeVarint(0, buffer); // empty headers
+ buffer.putLong(timestamp);
+ buffer.put(plainValue);
+ return buffer.array();
+ }
+
+ // Strips the ValueTimestampHeaders wrapper ([headersSize][headers][timestamp]) from a value
+ // part, leaving the plain value bytes. Used to normalize restored V4 records to the in-memory
+ // encoding used when header stores are disabled.
+ private static byte[] unwrapHeadersFormatToPlainValue(final byte[] headersFormatValue) {
+ if (headersFormatValue == null) {
+ return null;
+ }
+ final ByteBuffer buffer = ByteBuffer.wrap(headersFormatValue);
+ final int headersSize = ByteUtils.readVarint(buffer);
+ buffer.position(buffer.position() + headersSize + Long.BYTES);
+ final byte[] plainValue = new byte[buffer.remaining()];
+ buffer.get(plainValue);
+ return plainValue;
+ }
+
@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 newHeaders = new RecordHeaders(record.headers());
+ final long newTimestamp = record.timestamp();
+ final Bytes serializedKey = Bytes.wrap(keySerde.serializer().serialize(changelogTopic, newHeaders, 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(
Review Comment:
Yes, but does the recordContext not already have the right headers for plain mode case? Why do we need to set it? -- So also not clear to me, why we need the change further below (seems I am missing something)?
```
- new BufferValue(serializedPriorValue, serialChange.oldValue, serialChange.newValue, recordContext)
+ new BufferValue(serializedPriorValue, oldValue, newValue, effectiveContext)
```
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-20413: Make Streams suppress headers-aware [kafka]
Posted by "mjsax (via GitHub)" <gi...@apache.org>.
mjsax commented on PR #22165:
URL: https://github.com/apache/kafka/pull/22165#issuecomment-5113949164
Thanks for the PR! -- What a piece of work! Congrats on landing this one.
Merged to trunk.
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-20413: Make Streams suppress headers-aware [kafka]
Posted by "aliehsaeedii (via GitHub)" <gi...@apache.org>.
aliehsaeedii commented on PR #22165:
URL: https://github.com/apache/kafka/pull/22165#issuecomment-5104122941
> I think we are close... -- one main thing is missing tests, but we can add them as follow up PRs. In particular, it would be good to have a downgrade system test (maybe also upgrade) to ensure the changes to the changelog topic don't crash 4.3 or older clients. -- I believe we already have some tests in general, so maybe it's just a matter of extending them (or if we are lucky, suppress() might already be covered).
`suppress()` is in the smoke-test topology (`SmokeTestClient#183,195`, both `untilWindowCloses`), that app is what
`streams_application_upgrade_test.py` drives.
Two gaps, though, so it's a bit more than extending what exists:
1. No system test sets `dsl.store.format` anywhere.
2. The harness is upgrade-only: `to_version` is hardcoded to `DEV_VERSION` and
`restart_all_nodes_with()` only moves nodes forward. The downgrade direction has to be built.
I'll file a ticket for it.
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-20413: Make Streams suppress headers-aware [kafka]
Posted by "aliehsaeedii (via GitHub)" <gi...@apache.org>.
aliehsaeedii commented on PR #22165:
URL: https://github.com/apache/kafka/pull/22165#issuecomment-4968386360
> The fix is surprisingly simple... Should we add some TTD test (or even integration test) for some e2e verification?
Good point. Added.
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-20413: Make Streams suppress headers-aware [kafka]
Posted by "aliehsaeedii (via GitHub)" <gi...@apache.org>.
aliehsaeedii commented on code in PR #22165:
URL: https://github.com/apache/kafka/pull/22165#discussion_r3638341684
##########
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:
How should we do that? Is there any wiring from the upstream store's header-awareness into the suppress buffer today?
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-20413: Make Streams suppress headers-aware [kafka]
Posted by "aliehsaeedii (via GitHub)" <gi...@apache.org>.
aliehsaeedii commented on code in PR #22165:
URL: https://github.com/apache/kafka/pull/22165#discussion_r3638418494
##########
streams/src/main/java/org/apache/kafka/streams/state/internals/InMemoryTimeOrderedKeyValueChangeBuffer.java:
##########
@@ -465,28 +513,165 @@ 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 ? unwrapHeadersFormatToPlainValue(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(
+ wrapPlainValueAsHeadersFormat(value.priorValue(), timestamp),
+ wrapPlainValueAsHeadersFormat(value.oldValue(), timestamp),
+ wrapPlainValueAsHeadersFormat(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(
Review Comment:
Ok, of course we can, But why???? we do not support downgrade fo rthe whole thing. Why should we do that in this small corner? it is confusing as is not consistent with the rest of the code and the KIPs. WDYT?
##########
streams/src/main/java/org/apache/kafka/streams/state/internals/InMemoryTimeOrderedKeyValueChangeBuffer.java:
##########
@@ -465,28 +513,165 @@ 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 ? unwrapHeadersFormatToPlainValue(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(
+ wrapPlainValueAsHeadersFormat(value.priorValue(), timestamp),
+ wrapPlainValueAsHeadersFormat(value.oldValue(), timestamp),
+ wrapPlainValueAsHeadersFormat(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(
Review Comment:
Ok, of course we can, But why? we do not support downgrade fo rthe whole thing. Why should we do that in this small corner? it is confusing as is not consistent with the rest of the code and the KIPs. WDYT?
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-20413: Make Streams suppress headers-aware [kafka]
Posted by "aliehsaeedii (via GitHub)" <gi...@apache.org>.
aliehsaeedii commented on code in PR #22165:
URL: https://github.com/apache/kafka/pull/22165#discussion_r3638376782
##########
streams/src/main/java/org/apache/kafka/streams/state/internals/InMemoryTimeOrderedKeyValueChangeBuffer.java:
##########
@@ -399,12 +438,12 @@ 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())
+ final Headers headers = bufferValue.context().headers();
+ final K key = keySerde.deserializer().deserialize(changelogTopic, headers, next.getKey().key().get());
+ final Change<V> value = new Change<>(
+ deserializeValue(bufferValue.newValue(), headers),
+ deserializeValue(bufferValue.oldValue(), headers)
Review Comment:
they do in headers mode (separate ValueTimestampHeaders blobs). Only plain mode shares one context (no per-value headers exist there).
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-20413: Make Streams suppress headers-aware [kafka]
Posted by "aliehsaeedii (via GitHub)" <gi...@apache.org>.
aliehsaeedii commented on code in PR #22165:
URL: https://github.com/apache/kafka/pull/22165#discussion_r3638400854
##########
streams/src/main/java/org/apache/kafka/streams/state/internals/InMemoryTimeOrderedKeyValueChangeBuffer.java:
##########
@@ -465,28 +513,165 @@ 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.
Review Comment:
it is stored as ValueTimestampHeaders in headers mode; plainPriorValueForBuffered only unwraps it for the legacy V0/V1 restore decoders that require plain bytes.
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-20413: Make Streams suppress headers-aware [kafka]
Posted by "mjsax (via GitHub)" <gi...@apache.org>.
mjsax commented on code in PR #22165:
URL: https://github.com/apache/kafka/pull/22165#discussion_r3591889688
##########
streams/src/test/java/org/apache/kafka/streams/kstream/internals/SuppressScenarioTest.java:
##########
@@ -161,6 +164,45 @@ public void shouldImmediatelyEmitEventsWithZeroEmitAfter() {
}
}
+ @Test
+ public void shouldPropagateHeadersThroughSuppression() {
+ final StreamsBuilder builder = new StreamsBuilder();
+
+ final KTable<String, Long> valueCounts = builder
+ .table(
+ "input",
+ Consumed.with(STRING_SERDE, STRING_SERDE),
+ Materialized.<String, String, KeyValueStore<Bytes, byte[]>>with(STRING_SERDE, STRING_SERDE)
+ .withCachingDisabled()
+ .withLoggingDisabled()
+ )
+ .groupBy((k, v) -> new KeyValue<>(v, k), Grouped.with(STRING_SERDE, STRING_SERDE))
+ .count();
+
+ valueCounts
+ .suppress(untilTimeLimit(ZERO, unbounded()))
+ .toStream()
+ .to("output-suppressed", Produced.with(STRING_SERDE, Serdes.Long()));
Review Comment:
I am also wondering, if we should add another `process()` step before we write to the output topic, to access `context.headers()` (and maybe add information about it to the record's value, so we can capture it on the output topic).
When the suppress() flush happens while processing the second record, the context should not have been set to the flushed/first record's header I believe, but should be the headers from the second record triggering the flush?
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-20413: Make Streams suppress headers-aware [kafka]
Posted by "mjsax (via GitHub)" <gi...@apache.org>.
mjsax commented on code in PR #22165:
URL: https://github.com/apache/kafka/pull/22165#discussion_r3626775223
##########
streams/src/main/java/org/apache/kafka/streams/state/internals/InMemoryTimeOrderedKeyValueChangeBuffer.java:
##########
@@ -399,12 +438,12 @@ 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())
+ final Headers headers = bufferValue.context().headers();
+ final K key = keySerde.deserializer().deserialize(changelogTopic, headers, next.getKey().key().get());
+ final Change<V> value = new Change<>(
+ deserializeValue(bufferValue.newValue(), headers),
+ deserializeValue(bufferValue.oldValue(), headers)
Review Comment:
old-value and new-value should have individual headers, right, not the same?
##########
streams/src/main/java/org/apache/kafka/streams/state/internals/InMemoryTimeOrderedKeyValueChangeBuffer.java:
##########
@@ -399,12 +438,12 @@ 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())
+ final Headers headers = bufferValue.context().headers();
Review Comment:
Why are we getting the context headers here? Should we not access `bufferValue.newValue().headers()` instead?
##########
streams/src/main/java/org/apache/kafka/streams/state/internals/InMemoryTimeOrderedKeyValueChangeBuffer.java:
##########
@@ -437,12 +476,21 @@ public void evictWhile(final Supplier<Boolean> predicate,
@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).
Review Comment:
> they are unknown/empty when the key was first buffered
Why? Each put into the buffer should be put with a full `Record` which should have a valid headers object?
##########
streams/src/main/java/org/apache/kafka/streams/state/internals/InMemoryTimeOrderedKeyValueChangeBuffer.java:
##########
@@ -465,28 +513,165 @@ 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 ? unwrapHeadersFormatToPlainValue(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(
+ wrapPlainValueAsHeadersFormat(value.priorValue(), timestamp),
+ wrapPlainValueAsHeadersFormat(value.oldValue(), timestamp),
+ wrapPlainValueAsHeadersFormat(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);
+ }
+
+ // Wraps a plain value part as a ValueTimestampHeaders blob ([headersSize=0][timestamp][value])
+ // without needing a value serde. Used to normalize restored V0-V3 records to the in-memory
+ // encoding used when header stores are enabled.
+ private static byte[] wrapPlainValueAsHeadersFormat(final byte[] plainValue, final long timestamp) {
+ if (plainValue == null) {
+ return null;
+ }
+ final ByteBuffer buffer = ByteBuffer.allocate(ByteUtils.sizeOfVarint(0) + Long.BYTES + plainValue.length);
+ ByteUtils.writeVarint(0, buffer); // empty headers
+ buffer.putLong(timestamp);
+ buffer.put(plainValue);
+ return buffer.array();
+ }
+
+ // Strips the ValueTimestampHeaders wrapper ([headersSize][headers][timestamp]) from a value
+ // part, leaving the plain value bytes. Used to normalize restored V4 records to the in-memory
+ // encoding used when header stores are disabled.
+ private static byte[] unwrapHeadersFormatToPlainValue(final byte[] headersFormatValue) {
Review Comment:
Should we move this into the `Utils` class we added for all such helpers?
##########
streams/src/main/java/org/apache/kafka/streams/state/internals/InMemoryTimeOrderedKeyValueChangeBuffer.java:
##########
@@ -465,28 +513,165 @@ 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.
Review Comment:
Not sure if I understand this -- why would priorValue not be `ValueTimestampeHeaders`, too?
##########
streams/src/main/java/org/apache/kafka/streams/state/internals/InMemoryTimeOrderedKeyValueChangeBuffer.java:
##########
@@ -465,28 +513,165 @@ 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 ? unwrapHeadersFormatToPlainValue(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(
+ wrapPlainValueAsHeadersFormat(value.priorValue(), timestamp),
+ wrapPlainValueAsHeadersFormat(value.oldValue(), timestamp),
+ wrapPlainValueAsHeadersFormat(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);
+ }
+
+ // Wraps a plain value part as a ValueTimestampHeaders blob ([headersSize=0][timestamp][value])
+ // without needing a value serde. Used to normalize restored V0-V3 records to the in-memory
+ // encoding used when header stores are enabled.
+ private static byte[] wrapPlainValueAsHeadersFormat(final byte[] plainValue, final long timestamp) {
+ if (plainValue == null) {
+ return null;
+ }
+ final ByteBuffer buffer = ByteBuffer.allocate(ByteUtils.sizeOfVarint(0) + Long.BYTES + plainValue.length);
+ ByteUtils.writeVarint(0, buffer); // empty headers
+ buffer.putLong(timestamp);
+ buffer.put(plainValue);
+ return buffer.array();
+ }
+
+ // Strips the ValueTimestampHeaders wrapper ([headersSize][headers][timestamp]) from a value
+ // part, leaving the plain value bytes. Used to normalize restored V4 records to the in-memory
+ // encoding used when header stores are disabled.
+ private static byte[] unwrapHeadersFormatToPlainValue(final byte[] headersFormatValue) {
+ if (headersFormatValue == null) {
+ return null;
+ }
+ final ByteBuffer buffer = ByteBuffer.wrap(headersFormatValue);
+ final int headersSize = ByteUtils.readVarint(buffer);
+ buffer.position(buffer.position() + headersSize + Long.BYTES);
+ final byte[] plainValue = new byte[buffer.remaining()];
+ buffer.get(plainValue);
+ return plainValue;
+ }
+
@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 newHeaders = new RecordHeaders(record.headers());
+ final long newTimestamp = record.timestamp();
+ final Bytes serializedKey = Bytes.wrap(keySerde.serializer().serialize(changelogTopic, newHeaders, 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(),
+ newHeaders
+ );
+
+ // 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.
Review Comment:
> on the first insert for a key they are genuinely unknown.
But for this case `oldValue == null` so why do we care? It doesn't seems to be a case that needs any special handling?
##########
streams/src/main/java/org/apache/kafka/streams/state/internals/InMemoryTimeOrderedKeyValueChangeBuffer.java:
##########
@@ -465,28 +513,165 @@ 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 ? unwrapHeadersFormatToPlainValue(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(
+ wrapPlainValueAsHeadersFormat(value.priorValue(), timestamp),
+ wrapPlainValueAsHeadersFormat(value.oldValue(), timestamp),
+ wrapPlainValueAsHeadersFormat(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);
+ }
+
+ // Wraps a plain value part as a ValueTimestampHeaders blob ([headersSize=0][timestamp][value])
+ // without needing a value serde. Used to normalize restored V0-V3 records to the in-memory
+ // encoding used when header stores are enabled.
+ private static byte[] wrapPlainValueAsHeadersFormat(final byte[] plainValue, final long timestamp) {
+ if (plainValue == null) {
+ return null;
+ }
+ final ByteBuffer buffer = ByteBuffer.allocate(ByteUtils.sizeOfVarint(0) + Long.BYTES + plainValue.length);
+ ByteUtils.writeVarint(0, buffer); // empty headers
+ buffer.putLong(timestamp);
+ buffer.put(plainValue);
+ return buffer.array();
+ }
+
+ // Strips the ValueTimestampHeaders wrapper ([headersSize][headers][timestamp]) from a value
+ // part, leaving the plain value bytes. Used to normalize restored V4 records to the in-memory
+ // encoding used when header stores are disabled.
+ private static byte[] unwrapHeadersFormatToPlainValue(final byte[] headersFormatValue) {
+ if (headersFormatValue == null) {
+ return null;
+ }
+ final ByteBuffer buffer = ByteBuffer.wrap(headersFormatValue);
+ final int headersSize = ByteUtils.readVarint(buffer);
+ buffer.position(buffer.position() + headersSize + Long.BYTES);
+ final byte[] plainValue = new byte[buffer.remaining()];
+ buffer.get(plainValue);
+ return plainValue;
+ }
+
@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 newHeaders = new RecordHeaders(record.headers());
+ final long newTimestamp = record.timestamp();
+ final Bytes serializedKey = Bytes.wrap(keySerde.serializer().serialize(changelogTopic, newHeaders, 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(
Review Comment:
I don't understand why we would need this? We store the header already in the value, so on evict, we should extract it from the value accordingly.
##########
streams/src/main/java/org/apache/kafka/streams/state/internals/InMemoryTimeOrderedKeyValueChangeBuffer.java:
##########
@@ -465,28 +513,165 @@ 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 ? unwrapHeadersFormatToPlainValue(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(
+ wrapPlainValueAsHeadersFormat(value.priorValue(), timestamp),
+ wrapPlainValueAsHeadersFormat(value.oldValue(), timestamp),
+ wrapPlainValueAsHeadersFormat(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);
+ }
+
+ // Wraps a plain value part as a ValueTimestampHeaders blob ([headersSize=0][timestamp][value])
+ // without needing a value serde. Used to normalize restored V0-V3 records to the in-memory
+ // encoding used when header stores are enabled.
+ private static byte[] wrapPlainValueAsHeadersFormat(final byte[] plainValue, final long timestamp) {
+ if (plainValue == null) {
+ return null;
+ }
+ final ByteBuffer buffer = ByteBuffer.allocate(ByteUtils.sizeOfVarint(0) + Long.BYTES + plainValue.length);
+ ByteUtils.writeVarint(0, buffer); // empty headers
+ buffer.putLong(timestamp);
+ buffer.put(plainValue);
+ return buffer.array();
+ }
+
+ // Strips the ValueTimestampHeaders wrapper ([headersSize][headers][timestamp]) from a value
+ // part, leaving the plain value bytes. Used to normalize restored V4 records to the in-memory
+ // encoding used when header stores are disabled.
+ private static byte[] unwrapHeadersFormatToPlainValue(final byte[] headersFormatValue) {
+ if (headersFormatValue == null) {
+ return null;
+ }
+ final ByteBuffer buffer = ByteBuffer.wrap(headersFormatValue);
+ final int headersSize = ByteUtils.readVarint(buffer);
+ buffer.position(buffer.position() + headersSize + Long.BYTES);
+ final byte[] plainValue = new byte[buffer.remaining()];
+ buffer.get(plainValue);
+ return plainValue;
+ }
+
@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 newHeaders = new RecordHeaders(record.headers());
+ final long newTimestamp = record.timestamp();
Review Comment:
```suggestion
final long timestamp = record.timestamp();
```
##########
streams/src/main/java/org/apache/kafka/streams/state/internals/InMemoryTimeOrderedKeyValueChangeBuffer.java:
##########
@@ -465,28 +513,165 @@ 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 ? unwrapHeadersFormatToPlainValue(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(
+ wrapPlainValueAsHeadersFormat(value.priorValue(), timestamp),
+ wrapPlainValueAsHeadersFormat(value.oldValue(), timestamp),
+ wrapPlainValueAsHeadersFormat(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);
+ }
+
+ // Wraps a plain value part as a ValueTimestampHeaders blob ([headersSize=0][timestamp][value])
+ // without needing a value serde. Used to normalize restored V0-V3 records to the in-memory
+ // encoding used when header stores are enabled.
+ private static byte[] wrapPlainValueAsHeadersFormat(final byte[] plainValue, final long timestamp) {
Review Comment:
Should we move this into the `Utils` class we added for all such helpers?
##########
streams/src/main/java/org/apache/kafka/streams/state/internals/InMemoryTimeOrderedKeyValueChangeBuffer.java:
##########
@@ -465,28 +513,165 @@ 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 ? unwrapHeadersFormatToPlainValue(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(
+ wrapPlainValueAsHeadersFormat(value.priorValue(), timestamp),
+ wrapPlainValueAsHeadersFormat(value.oldValue(), timestamp),
+ wrapPlainValueAsHeadersFormat(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);
+ }
+
+ // Wraps a plain value part as a ValueTimestampHeaders blob ([headersSize=0][timestamp][value])
+ // without needing a value serde. Used to normalize restored V0-V3 records to the in-memory
+ // encoding used when header stores are enabled.
+ private static byte[] wrapPlainValueAsHeadersFormat(final byte[] plainValue, final long timestamp) {
+ if (plainValue == null) {
+ return null;
+ }
+ final ByteBuffer buffer = ByteBuffer.allocate(ByteUtils.sizeOfVarint(0) + Long.BYTES + plainValue.length);
+ ByteUtils.writeVarint(0, buffer); // empty headers
+ buffer.putLong(timestamp);
+ buffer.put(plainValue);
+ return buffer.array();
+ }
+
+ // Strips the ValueTimestampHeaders wrapper ([headersSize][headers][timestamp]) from a value
+ // part, leaving the plain value bytes. Used to normalize restored V4 records to the in-memory
+ // encoding used when header stores are disabled.
+ private static byte[] unwrapHeadersFormatToPlainValue(final byte[] headersFormatValue) {
+ if (headersFormatValue == null) {
+ return null;
+ }
+ final ByteBuffer buffer = ByteBuffer.wrap(headersFormatValue);
+ final int headersSize = ByteUtils.readVarint(buffer);
+ buffer.position(buffer.position() + headersSize + Long.BYTES);
+ final byte[] plainValue = new byte[buffer.remaining()];
+ buffer.get(plainValue);
+ return plainValue;
+ }
+
@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 newHeaders = new RecordHeaders(record.headers());
Review Comment:
```suggestion
final RecordHeaders headers = new RecordHeaders(record.headers());
```
We also call it `key` and `value`, not `newKey` or `newValue`?
##########
streams/src/main/java/org/apache/kafka/streams/state/internals/InMemoryTimeOrderedKeyValueChangeBuffer.java:
##########
@@ -465,28 +513,165 @@ 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 ? unwrapHeadersFormatToPlainValue(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(
+ wrapPlainValueAsHeadersFormat(value.priorValue(), timestamp),
+ wrapPlainValueAsHeadersFormat(value.oldValue(), timestamp),
+ wrapPlainValueAsHeadersFormat(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(
Review Comment:
I think we can easily support this actually, because `suppress()` is in-memory, so there is no local state-store concern.
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-20413: Make Streams suppress headers-aware [kafka]
Posted by "aliehsaeedii (via GitHub)" <gi...@apache.org>.
aliehsaeedii commented on code in PR #22165:
URL: https://github.com/apache/kafka/pull/22165#discussion_r3638697901
##########
streams/src/main/java/org/apache/kafka/streams/state/internals/InMemoryTimeOrderedKeyValueChangeBuffer.java:
##########
@@ -437,12 +476,21 @@ public void evictWhile(final Supplier<Boolean> predicate,
@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).
Review Comment:
you are right but those are the new value's headers, which we do capture. The "unknown/empty" is only the prior value at first buffering, whose headers were never delivered across the Change boundary.
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-20413: Make Streams suppress headers-aware [kafka]
Posted by "mjsax (via GitHub)" <gi...@apache.org>.
mjsax commented on code in PR #22165:
URL: https://github.com/apache/kafka/pull/22165#discussion_r3591106823
##########
streams/src/test/java/org/apache/kafka/streams/kstream/internals/SuppressScenarioTest.java:
##########
@@ -161,6 +164,45 @@ public void shouldImmediatelyEmitEventsWithZeroEmitAfter() {
}
}
+ @Test
+ public void shouldPropagateHeadersThroughSuppression() {
+ final StreamsBuilder builder = new StreamsBuilder();
+
+ final KTable<String, Long> valueCounts = builder
+ .table(
Review Comment:
Not sure why we read the input as table? Seem unnecessary complex if we chain a `count()`? Or we just omit the `count()`?
##########
streams/src/test/java/org/apache/kafka/streams/kstream/internals/SuppressScenarioTest.java:
##########
@@ -161,6 +164,45 @@ public void shouldImmediatelyEmitEventsWithZeroEmitAfter() {
}
}
+ @Test
+ public void shouldPropagateHeadersThroughSuppression() {
+ final StreamsBuilder builder = new StreamsBuilder();
+
+ final KTable<String, Long> valueCounts = builder
+ .table(
+ "input",
+ Consumed.with(STRING_SERDE, STRING_SERDE),
+ Materialized.<String, String, KeyValueStore<Bytes, byte[]>>with(STRING_SERDE, STRING_SERDE)
+ .withCachingDisabled()
+ .withLoggingDisabled()
+ )
+ .groupBy((k, v) -> new KeyValue<>(v, k), Grouped.with(STRING_SERDE, STRING_SERDE))
+ .count();
+
+ valueCounts
+ .suppress(untilTimeLimit(ZERO, unbounded()))
Review Comment:
I don't think this work correctly. If we set `ZERO` we flush the suppress buffer right away, and even in the old code, headers would get carried forward via the context... (I just ran this this unmodified on `trunk` and it also passed).
Instead, we need to set an actual suppress time and pipe two input record. The second record (with larger ts) would flush the suppress buffer emitting the first. W/o the first, the headers from the second record would carry forward into the result, but with the fix, the correct headers from the suppressed record should be there.
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-20413: Make Streams suppress headers-aware [kafka]
Posted by "mjsax (via GitHub)" <gi...@apache.org>.
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-20413: Make Streams suppress headers-aware [kafka]
Posted by "mjsax (via GitHub)" <gi...@apache.org>.
mjsax commented on code in PR #22165:
URL: https://github.com/apache/kafka/pull/22165#discussion_r3653553371
##########
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:
Not sure -- guess we could also omit this for now, but it seems to be thing uses would need to know about, and we should document clearly -- I am not sure if users would expect it, and w/o a switch on `Suppress` to turn it on/off it would be difficult to turn it on for only a single KTable plus suppress() [even if this might be a rare use-case, so acceptable for now].
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-20413: Make Streams suppress headers-aware [kafka]
Posted by "mjsax (via GitHub)" <gi...@apache.org>.
mjsax commented on code in PR #22165:
URL: https://github.com/apache/kafka/pull/22165#discussion_r3653581797
##########
streams/src/main/java/org/apache/kafka/streams/state/internals/InMemoryTimeOrderedKeyValueChangeBuffer.java:
##########
@@ -465,28 +513,165 @@ 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 ? unwrapHeadersFormatToPlainValue(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(
+ wrapPlainValueAsHeadersFormat(value.priorValue(), timestamp),
+ wrapPlainValueAsHeadersFormat(value.oldValue(), timestamp),
+ wrapPlainValueAsHeadersFormat(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(
Review Comment:
We don't support rolling bounce downgrade, but a downgrade via app shutdown + wipe local state directory + restart with old version. And for `suppress()` the "wipe local state directory" is jus a no-op.
It seems, we would crash though with this exception if one tries this downgrade path that we do support? So why disallow it?
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-20413: Make Streams suppress headers-aware [kafka]
Posted by "aliehsaeedii (via GitHub)" <gi...@apache.org>.
aliehsaeedii commented on code in PR #22165:
URL: https://github.com/apache/kafka/pull/22165#discussion_r3558729010
##########
streams/src/main/java/org/apache/kafka/streams/state/internals/InMemoryTimeOrderedKeyValueChangeBuffer.java:
##########
@@ -437,19 +436,22 @@ public void evictWhile(final Supplier<Boolean> predicate,
@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();
final V deserializedValue = valueSerde.innerSerde().deserializer().deserialize(
changelogTopic,
- context.headers(),
+ bufferValue.context().headers(),
serializedValue
);
- // it's unfortunately not possible to know this, unless we materialize the suppressed result, since our only
- // knowledge of the prior value is what the upstream processor sends us as the "old value" when we first
- // buffer something.
- return Maybe.defined(ValueTimestampHeaders.make(deserializedValue, RecordQueue.UNKNOWN, new RecordHeaders()));
+ return Maybe.defined(ValueTimestampHeaders.make(
+ deserializedValue,
+ bufferValue.context().timestamp(),
Review Comment:
Due to these (above inline comments) explanations I reverted the changes of here. The former [commit](https://github.com/apache/kafka/pull/22165/changes/87bee53184f012260b5bad0c293e653a3601c7e9#diff-b94df16c1e8d51db85b6539dce34da2e6cfda0a42584607d60c867e543f268e7) returns `bufferValue.context().timestamp()` instead of `RecordQueue.UNKNOWN` and `RecordHeaders(bufferValue.context().headers())` instead of empty headers.
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-20413: Make Streams suppress headers-aware [kafka]
Posted by "aliehsaeedii (via GitHub)" <gi...@apache.org>.
aliehsaeedii commented on code in PR #22165:
URL: https://github.com/apache/kafka/pull/22165#discussion_r3664770759
##########
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:
Good point. We generally must update the docs after this PR: https://issues.apache.org/jira/browse/KAFKA-20849
>Should we file a Jira ticket about "allow enabling header-support on suppress() similar to Materialized for other operators"?
Mmm. Does it need a KIP? does it override dsl.store.format? I don't fully understand this.
##########
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:
"Part" means prior/old/new. The three value parts of a BufferValue — but the method itself knows nothing about that. it's just raw VTH bytes → ValueTimestampHeaders. So the name describes the caller, not the callee. I rename it to `deserializeValueTimestampHeaders`
##########
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:
It is a Header object whose value is the [headersSize][headers][timestamp] byte prefix. I renamed it to
`headersAndTimestamp`
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org