Major Description
Using Kafka 0.10.0.1, if joining records in two streams separated by some time, but only when records from one stream are newer than records from the other, i.e. doing:
stream1.join(stream2, valueJoiner, JoinWindows.of("X").after(10000))
One would expect that the following would be equivalent:
stream2.join(stream1, valueJoiner, JoinWindows.of("X").before(10000))
Alas, that this is not the case. Instead, this generates the same output as the first example:
stream2.join(stream1, valueJoiner, JoinWindows.of("X").after(10000))
The problem is that the DefaultJoin implementation in KStreamImpl fails to reverse the before and after values when creates the KStreamKStreamJoin for the other stream, even though is calls reverseJoiner to reverse the joiner.
Attachments
Issue Links
- links to
