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