KAFKA-6521: Use timestamped stores for KTables

[mjsax]

The tests are failing because #6661 is not merged yet..

The goal of this PR is to swap out, KeyValueStore and WindowStore with the corresponding new TimestampedXxxStore in the DSL where appropriate. We don't leverage the timestamps yet in the Processors, to keep the PR contained. Thus, no semantic change yet.

[mjsax]

Call for early review @guozhangwang @bbejeck @vvcephei @ableegoldman @cadonna @abbccdda

[mjsax]

Fix bug that was exposed with this PR.

[vvcephei]

Oof. Good catch!

[vvcephei]

Thanks for the PR, @mjsax . I guess the idea here is to just mix in the new (timestamped) stores without changing any semantics? And then we'll have other PR(s) later to update the semantics and tests? This seems like a good way to break it down.

Just a few comments.

[vvcephei]

This logic has come up once or twice ;)

Just a random idea... What do you think about a static method on ValueAndTimestamp:

static <V> V getValueOrNull(final ValueAndTimestamp<V> valueAndTimestamp) {
  return valueAndTimestamp == null ? null : valueAndTimestamp.value();
}

Then all these other methods could be:

Suggested change

	return valueAndTimestamp == null ? null : valueAndTimestamp.value();
	return getValueOrNull(valueAndTimestamp);

?

[mjsax]

Ack. I'll just update the KIP accordingly. Turn out, we already have ValueAndTimestamp#make and no public constructor (also not documented on the KIP).

[vvcephei]

Oof. Good catch!

[vvcephei]

I found it useful with the FullChangeSerde to make the constructor private and force usage via a null-safe factory method. Then, you never have to worry about this check.

[mjsax]

Seems to be a single occurrence. Maybe not worth a factory. Also, ValueAndTimestampSerde checks for non-null and thus we would surface any issue during testing anyway.

[mjsax]

Updated this.

[vvcephei]

Thanks for the update!

Just another minor comment on ValueAndTimestamp

[vvcephei]

Since the only usage of the constructor is in this class, it should be ok to be private.

Also, since it's just a data struct, we could also make the class final.

[mjsax]

Updated this.

[abbccdda]

We already import the class on L26, so let's remove this import.

[guozhangwang]

This is for statically use this single util function (otherwise we have to call ValueAndTimestamp.getValueOrNull when calling).

[abbccdda]

Same here.

[abbccdda]

Is having curly braces in one line a convention?

[abbccdda]

Seems unnecessary L123

[mjsax]

I like to have some more line to separate nested classes from method

[abbccdda]

We could omit L28-29 since we don't want to comment on it.

[vvcephei]

Good eye. It would be even better to document the parameters!

[abbccdda]

How about

if (!cachingEnabled) {
    context.forward(key, new Change<>(newValue, oldValue));
}

[abbccdda]

s/is not/if not

[bbejeck]

#6661 has been merged now.

retest this please

[bbejeck]

Thanks @mjsax LGTM

[guozhangwang]

retest this please

[guozhangwang]

Some minor comments.

[guozhangwang]

This is for statically use this single util function (otherwise we have to call ValueAndTimestamp.getValueOrNull when calling).

[guozhangwang]

Today we store the timestamp in cache entry and upon flushing we call entry.entry().context().timestamp() to set the timestamp. Would that be replaced with the underlying store's timestamp then?

[mjsax]

Maybe. After the store refactoring, I realized that we might actually be able to revert #6147 -- however, I would prefer to do this as cleanup only after everything else is merged (to make sure it's really safe).

[mjsax]

While working on a follow PR, I figure out the details. The TimestampedForwardingCacheFlushListener needs to extract the timestamp from the value, and set it as record timestamp. Similar to SessionCacheFlushListener in another PR: https://github.com/apache/kafka/pull/6645/files#diff-4039efe54e3b87d65c20bd7d14c1b149R43

[guozhangwang]

nit: we can still use the ternary operator here.

[guozhangwang]

Can we just remove the old CacheFlushListener / TupleForwarder since they are internal classes only used in the DSL?

[mjsax]

We still need the old one, because we don't use TimestampedSessionWindowStore in the DSL but keep SessionWindowStore.

[guozhangwang]

Can we replace other uses of GenericInMemoryKeyValueStore so we can remove the old class as well?

[mjsax]

Not sure. We don't remove "plain" stores, and thus it seems to make sense to keep existing tests and only rewrite DSL related tests?

[mjsax]

Updated and rebased this.

[mjsax]

This fixes a bug that was exposed after rebasing. We need to treat in-memory stores always as TimestampedByteStores to guarantee correct restore. (Similar below for window stores.)

[mjsax]

Added a new test to #6179 that also exposed this bug (cf. StoreUpgradeIntegrationTest)

[bbejeck]

I agree with the fix, but it raises another question for me.

I admit I could be either missing something or forgetting some details already spelled out. Why don't we convert the values for in-memory stores to also have timestamps in the value? I'm basing on the fact that the InMemoryTimestampedKeyValueStoreMarker does simple puts/gets.

[mjsax]

The in-memory store will have no content on startup, and the upper layer will put/get byte[] -- the store does not care what is in it. The upper layer will already serialize value+timestamp and put those, and expects value+timestamp byte[] back -- hence, the in-memory store does not need to do any translation.

For persistent store, the store does have content on local disk, namely plain values. Because this store cannot be upgraded, we put the adapter in between, to remove the timestamp part on put() and add dummy timestamp on get(), to meet the contract to the upper layer that expects byte[] to be value+timestamp.

If we restore a persistent store, the just copy plain key/value from the changelog topic and put it into the store. Everything still works as expected.

If we restore an in-memory store, we need to add the timestamp to the value, and this is triggered by the TimestampeByteStore interface. Otherwise, the in-memory store would contain plain values, and violate the contract to the upper layer, because the assumption is, that everything store in it, has value+timestamp format.

Ie, for persistent store, we play the backward compatibility game using surrogate timestamps. However, for in-memory store, we don't need to do this, because there is no local on-disk content and we can switch to timestamped-store behavior without any drawbacks and get full feature set.

Does this make sense?

[bbejeck]

yep thanks for clarifying

[mjsax]

"exposed" after rebase -- for time-windows, we want to use the end-timestamp (even if this change is not implemented yet, I added it here, to avoid touching this class later on again).

[mjsax]

The new wrapping would break the test. Using RocksDBTimestampedStore avoid the wrapping as TimestampedByteStore interface is implemented. (Same below.)

[mjsax]

Rebased to resolve merge conflicts.

[mjsax]

Minor fix on-the-side to not use deprecated method any longer.

[mjsax]

New fix: On flush, we take the timestamp from the newValue if it exists.

[mjsax]

New fix: TupleForwarder should obey sendOldValues, too.

[mjsax]

For testing...

[mjsax]

New: slipped in the first commit of this PR... discovered while working on a follow up PR.

[mjsax]

Rebased to resolve merge conflict. Also added some more fixes and tests.

[guozhangwang]

LGTM. Please feel free to merge after green builds.

[guozhangwang]

retest this please