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