guozhangwang commented on a change in pull request #11581:
URL: https://github.com/apache/kafka/pull/11581#discussion_r766899186
##########
File path:
streams/src/main/java/org/apache/kafka/streams/state/internals/CachingKeyValueStore.java
##########
@@ -107,10 +107,10 @@ private void putAndMaybeForward(final
ThreadCache.DirtyEntry entry,
// we can skip flushing to downstream as well as writing to
underlying store
if (rawNewValue != null || rawOldValue != null) {
// we need to get the old values if needed, and then put to
store, and then flush
- wrapped().put(entry.key(), entry.newValue());
-
final ProcessorRecordContext current = context.recordContext();
context.setRecordContext(entry.entry().context());
+ wrapped().put(entry.key(), entry.newValue());
Review comment:
I think there's some rationales to not set the processor context for the
evicted record, e.g. let's say we first put record A offset 0 into the cache,
and then record B offset 1, and then record C offset 2, the third insert caused
the cache to evict A, which will then be written to the underlying store and
also be forwarded downstreams, if we set the metadata as offset `2` for record
`A` that may not be correct --- but I admit that if we do not set it, then it
would be either some order context like offset `1` or even just a null, but it
seems forwarding `A` with offset `2` is not appropriate still since we would
prefer to forward `A` with offset `0` ideally
--
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: jira-unsubscr...@kafka.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
