vcrfxia commented on code in PR #13564:
URL: https://github.com/apache/kafka/pull/13564#discussion_r1171930596
##########
streams/src/main/java/org/apache/kafka/streams/kstream/internals/KTableKTableInnerJoin.java:
##########
@@ -139,7 +139,7 @@ public void process(final Record<K, Change<V1>> record) {
oldValue = joiner.apply(record.value().oldValue, valueRight);
}
- context().forward(record.withValue(new Change<>(newValue,
oldValue)).withTimestamp(resultTimestamp));
+ context().forward(record.withValue(new Change<>(newValue,
oldValue, record.value().isLatest)).withTimestamp(resultTimestamp));
Review Comment:
Coming back to your question on this PR, yes `isLatest` will always be true
in this join processor, but it's not necessarily the case in the join merge
processor. It would not necessarily be true in the case where a user
materialized the join result as a versioned store (even though they shouldn't,
semantically), if they encounter this edge case with nulls.
##########
streams/src/main/java/org/apache/kafka/streams/kstream/internals/KTableKTableJoinMerger.java:
##########
@@ -117,13 +120,16 @@ public void init(final ProcessorContext<K, Change<V>>
context) {
@Override
public void process(final Record<K, Change<V>> record) {
if (queryableName != null) {
- store.put(record.key(), record.value().newValue,
record.timestamp());
- tupleForwarder.maybeForward(record);
+ final long putReturnCode = store.put(record.key(),
record.value().newValue, record.timestamp());
+ // if not put to store, do not forward downstream either
Review Comment:
Accidentally responded above in
https://github.com/apache/kafka/pull/13564#discussion_r1171930596, instead of
here.
--
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
