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