KIP-557: Add emit on change support to Kafka Streams

[ConcurrencyPractitioner]

This is a continuation of the previous PR #8254, and will move to implement emit on change support more extensively and cover most basic KTable operations.

[ConcurrencyPractitioner]

@vvcephei Just wanted to get your opinion on this. When adding emit-on-change support to KTableFilter, I found that the old value was already available for us in certain situations.
I thought that in this case, it would be wasteful if we just queried the store only to retrieve the timestamp. To that end, I decided that the value for serializedOldValue will simply just be ValueAndTimestamp(oldValue, 0). My reasoning is that the old value is guaranteed to have a smaller timestamp than the new value. So any value less than the current timestamp would suffice. I chose 0 just to be safe. Do you think this is safe?

[ConcurrencyPractitioner]

@vvcephei Think you have time to look at this?

[vvcephei]

Hey @ConcurrencyPractitioner , I'm sorry for the delay. I started to look at it, but got caught up in stabilizing the 2.6.0 and 2.5.1 releases. I'll get you a review ASAP.

[vvcephei]

Hey again @ConcurrencyPractitioner . I'm sorry to say that I still haven't been able to review. It was a long week last week, resolving a subtle and complex bug around upgrading with Suppression enabled.

I did familiarize myself with the issue you pointed out about this PR, and it's a real bummer. I think when I was initially considering how we could implement this without needing any additional serialization calls or lookups, I wasn't thinking about including timestamp semantics in the feature.

Unfortunately, it is possible for the "old" value to have a higher timestamp than the current record. If you recall, this was the reason to add that weird compareValuesAndCheckForIncreasingTimestamp in the last PR.

I think the right approach here would be to do a re-write of the internals so that we actually get the old timestamp forwarded as well, but I do not think we should do that until KIP-478 is complete. Otherwise, the risk is too high. The good news is that I've got a good chunk of that KIP complete already, and I'm just waiting to submit the PR until all the release craziness is done.

In the mean time, perhaps we can just focus on the operators for which this isn't a problem. For example, I just looked at KTableAggregate. In that processor, we have to load the prior value AND timestamp already.

WDYT about just leaving this PR open for now, and switching to focus on the processors that already load the prior value and timestamp? I'll tag you also on the KIP-478 PRs, so that you can help move that along and unblock yourself, if you like.

[ConcurrencyPractitioner]

@vvcephei Thanks for the thorough breakdown on what's going on! :) I will take a look at the other operators then. :)