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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
