aliehsaeedii commented on code in PR #22165:
URL: https://github.com/apache/kafka/pull/22165#discussion_r3638697901
##########
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:
you are right but those are the new value's headers, which we do capture.
The "unknown/empty" is only the prior value at first buffering, whose headers
were never delivered across the Change boundary.
--
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]
