mjsax commented on code in PR #13496:
URL: https://github.com/apache/kafka/pull/13496#discussion_r1158814912
##########
streams/src/main/java/org/apache/kafka/streams/kstream/internals/KTableKTableInnerJoin.java:
##########
@@ -153,11 +154,37 @@ public void init(final ProcessorContext<?, ?> context) {
@Override
public ValueAndTimestamp<VOut> get(final K key) {
- final ValueAndTimestamp<V1> valueAndTimestamp1 =
valueGetter1.get(key);
+ return computeJoin(key, valueGetter1::get, valueGetter2::get);
+ }
+
+ @Override
+ public ValueAndTimestamp<VOut> get(final K key, final long
asOfTimestamp) {
Review Comment:
Wondering if this is correct? This "ValueGetter" queries the
(non-materialized) result table of in inner-join, and re-computes the result
on-the-fly? Maybe it will get clear in a follow up PR, but would love to get a
short / high-level explanation why it's the right thing to do (as I right now
don't understand it).
What I am worried about it the case, that we don't update the result for
out-of-order updates, but when we re-compute the result on the fly, we might
re-compute an "updated" result, and thus don't return the same record as
perviously emitted.
You call this out in the second example of the "Proposal" section on the KIP:
```
A: (timestamp = 0, value = a0)
B: (timestamp = 2, value = b2) -> latest record on B side, emits (a0, b2)
with timestamp=2
A: (timestamp = 5, value = a5) -> latest record on A side, emits (a5, b2)
with timestamp=5
A: (timestamp = 1, value = a1) -> out-of-order record on A side, no new join
result emitted. the older (a0, b2) join result is no longer correct.
```
We don't emit anything for the last record, while the older join result
becomes invalid (that's ok); but if we call `get(key,1)` later, would we get
back the older (not updated) `(a2,b0)` (which would be correct to return from a
consistency point of view because the value-getter should return the same thing
as if the result state-store would have been materialized, and the value-getter
is not in use), or would we get back a now updated result `(a1,b2)`?
--
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
