mjsax commented on code in PR #22156:
URL: https://github.com/apache/kafka/pull/22156#discussion_r3365945941
##########
streams/src/main/java/org/apache/kafka/streams/state/internals/ChangeLoggingListValueBytesStore.java:
##########
@@ -33,9 +32,9 @@ public void put(final Bytes key, final byte[] value) {
// we need to log the full new list and thus call get() on the inner
store below
// if the value is a tombstone, we delete the whole list and thus can
save the get call
if (value == null) {
- log(key, null, internalContext.recordContext().timestamp(), new
RecordHeaders());
+ log(key, null, internalContext.recordContext().timestamp(),
internalContext.recordContext().headers());
} else {
- log(key, wrapped().get(key),
internalContext.recordContext().timestamp(), new RecordHeaders());
+ log(key, wrapped().get(key),
internalContext.recordContext().timestamp(),
internalContext.recordContext().headers());
Review Comment:
Not sure if this is correct? Don't we need a
`ChangeLoggingListValueByteStoreWithHeader` class and get the headers from the
`valueAndHeaders` bytes array?
Similar to what we do with
`ChangeLoggingTimestampedKeyValueBytesStoreWithHeaders` ?
--
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]
