KAFKA-9929: Support backward iterator on SessionStore #9139

jeqo · 2020-08-07T16:24:09Z

Depends on #9138

More detailed description of your change,
if necessary. The PR title and PR message become
the squashed commit message, so use a separate
comment to ping reviewers.

Summary of testing strategy (including rationale)
for the feature or bug fix. Unit and/or integration
tests are expected for any behaviour change and
system tests should be considered for larger changes.

Committer Checklist (excluded from commit message)

Verify design and implementation
Verify test coverage and CI build status
Verify documentation (including upgrade notes)

jeqo · 2020-09-11T11:44:12Z

cc @ableegoldman and team, this should be ready to review.
Would also be great to look #9228 as part of KIP-666, to get the APIs aligned.

Thanks!

ableegoldman

Hey @jeqo -- finally got around to giving this a first pass, thanks for being so patient 😄

A few minor notes, mostly things we've covered in the KV/window store PRs already. I'm still going through the tests but things look good overall

streams/src/main/java/org/apache/kafka/streams/state/internals/CachingSessionStore.java

ableegoldman · 2020-10-06T04:27:12Z

streams/src/main/java/org/apache/kafka/streams/state/internals/CachingSessionStore.java

@@ -359,7 +431,11 @@ private void getNextSegmentIterator() {
            setCacheKeyRange(currentSegmentBeginTime(), currentSegmentLastTime());

            current.close();
-            current = context.cache().range(cacheName, cacheKeyFrom, cacheKeyTo);
+            if (forward) {


I think we're going to need some additional changes in this class similar to what we had in CachingWindowStore. Definitely at least in getNextSegmentIterator(). Let's make sure to have some cross-segment test coverage here as well, especially because the iteration logic of session store range queries is the hardest to wrap your head around out of all the stores (at least, it is for me)

One of the reasons these missing pieces are not throwing exceptions seems to be that Caching Store tests are only covering or InMemory or Persistent underlying stores. This means only one path of the caching logic is followed (e.g. persistent() ? iteratorWraper : otherIterator)

I rename the test classes to be explicit about what backing store is covered. We could probably cover this as another issue?

Ohh yeah good catch. Kind of weird that we're inconsistent with which underlying store type is used in the existing caching store tests, too. But thanks for adding the in-memory + persistent flavors for the SessionStore, I think it sounds reasonable to file a ticket to follow up on the Window/KV flavors and clean up the caching tests

streams/src/main/java/org/apache/kafka/streams/state/internals/InMemorySessionStore.java

ableegoldman · 2020-10-06T23:50:49Z

streams/src/main/java/org/apache/kafka/streams/state/internals/CachingSessionStore.java

            this.keyFrom = keyFrom;
            this.keyTo = keyTo;
            this.latestSessionStartTime = latestSessionStartTime;
            this.lastSegmentId = cacheFunction.segmentId(maxObservedTimestamp);
            this.segmentInterval = cacheFunction.getSegmentInterval();
+            this.forward = forward;

            this.currentSegmentId = cacheFunction.segmentId(earliestSessionEndTime);


Ok I think that for the reverse case, this should be initialized to cacheFunction.segmentId(maxObservedTimestamp) and lastSegmentId should be initialized to this (segmentId(earliestSessionEndTime)).

right! great catch! forgot to align this with WindowStore.

ableegoldman

Just a few more minor things, should definitely be able to merge this by feature freeze 👍

streams/src/test/java/org/apache/kafka/streams/state/internals/CachingSessionStoreTest.java

ableegoldman · 2020-10-07T00:18:41Z

streams/src/test/java/org/apache/kafka/test/ReadOnlySessionStoreStub.java

+    public KeyValueIterator<Windowed<K>, V> backwardFetch(K from, K to) {
+        return null;


I guess it probably doesn't matter since we presumably aren't using these backward methods of the ReadOnlySessionStoreStub, but it seems like it might result in some tricky NPEs to debug if ever someone does try to use it in a test. If you don't feel like implementing it I think it's fine to just throw UnsupportedOperationException and say that you'll have to implement this to use it.

Or just copy the code from the forward direction and flip it 🤷‍♀️ Same goes for all the methods in here that return null

mixed choice here. hope is good enough

ableegoldman · 2020-10-07T23:12:07Z

Retest this please

ableegoldman

LGTM! Thanks for the reverse iterator KIP 😄

ableegoldman · 2020-10-07T23:14:40Z

test this please

ableegoldman · 2020-10-07T23:40:58Z

Alright Jenkins is being wonky yet again, but I checked out this branch and ran the tests locally to verify.

vvcephei

Thanks, @jeqo !

I had a few complaints about formatting and one about the exception message. Since we're down to the limit, I'll go ahead and add a commit addressing these concerns, test it for myself (thanks for your efforts there, @ableegoldman ), and merge it.

Thanks so much for this PR.

vvcephei · 2020-10-07T15:30:42Z

streams/src/main/java/org/apache/kafka/streams/state/ReadOnlySessionStore.java

+    default KeyValueIterator<Windowed<K>, AGG> findSessions(final K key,
+                                                            final long earliestSessionEndTime,
+                                                            final long latestSessionStartTime) {
+        throw new UnsupportedOperationException("Moved from SessionStore");


It won't matter to users whether this method was moved from another interface or not. They just need to know why they're getting the exception. I.e., we just need to tell them that the store implementation they selected didn't implement the method.

Suggested change

throw new UnsupportedOperationException("Moved from SessionStore");

throw new UnsupportedOperationException("This API is not supported by this implementation of ReadOnlySessionStore.");

We should say the exact same thing in all default implementations. Right now, they're a bit inconsistent.

vvcephei · 2020-10-07T15:33:29Z

streams/src/main/java/org/apache/kafka/streams/state/internals/CachingSessionStore.java

-                        cacheFunction.cacheKey(keySchema.lowerRangeFixedSize(key, earliestSessionEndTime)),
-                        cacheFunction.cacheKey(keySchema.upperRangeFixedSize(key, latestSessionStartTime))
+                cacheFunction.cacheKey(keySchema.lowerRangeFixedSize(key, earliestSessionEndTime)),
+                cacheFunction.cacheKey(keySchema.upperRangeFixedSize(key, latestSessionStartTime))


There are a lot of unnecessary whitespace changes in this PR. You don't need to back them all out right now, but in the future, please clean up the diff before submitting a PR. These extra changes make it harder to review.

vvcephei · 2020-10-07T22:43:26Z

streams/src/main/java/org/apache/kafka/streams/state/internals/InMemorySessionStore.java

+                    if (forward) entries = nextKeyEntry.getValue().descendingMap().entrySet();
+                    else entries = nextKeyEntry.getValue().entrySet();


The code style discourages inline conditionals. It's more maintainable to always use blocks.

vvcephei · 2020-10-08T03:46:56Z

Ok, I've pushed a couple of commits addressing my feedback.

I absolutely share the desire to clean up bad formatting in the codebase, but when PRs are this extensive, I'd suggest eliminating absolutely all changes that aren't directly related to the change. The extra whitespace changes, etc., just add noise that makes it harder for reviewers to do their job. Plus, it increases the probability of merge conflicts.

I'd encourage sending a separate PR applying formatting changes or other style fixes. This is what I do myself.

Anyway, I rolled back the "extra" stuff. I also applied a couple of extra formatting changes to make the new code itself comply with the style guidelines.

And it looks like Jenkins finally woke up, so I'll let it go ahead and finish running.

vvcephei · 2020-10-08T04:19:54Z

Oh, ironically, there is a merge conflict with trunk, and it's due to one of the only "extra" changes I left in: renaming CachingStoreTest to CachingStoreTest and in the Session variant, making context a local variable.

I didn't un-rename that class, but I did restore context as a field. I copied over the newly-added test in trunk just to the InMemory variants of the tests. They only mock the inner store and test the init method anyway.

vvcephei · 2020-10-08T04:20:30Z

Jenkins is running again, so I'll let it finish and merge in the morning.

vvcephei · 2020-10-08T13:09:03Z

Ok, it's merged. I'm waiting on permission from Bill before cherry-picking to 2.7

Implements KIP-617 for `SessionStore` Reviewers: A. Sophie Blee-Goldman <[email protected]>, John Roesler <[email protected]>

vvcephei · 2020-10-08T13:22:53Z

cherry-picked to 2.7. Thanks @jeqo !

Implements KIP-617 for `SessionStore` Reviewers: A. Sophie Blee-Goldman <[email protected]>, John Roesler <[email protected]>

@chia7712

* Updating trunk versions after cutting branch for 2.7 * KAFKA-9929: Support backward iterator on SessionStore (apache#9139) Implements KIP-617 for `SessionStore` Reviewers: A. Sophie Blee-Goldman <[email protected]>, John Roesler <[email protected]> * MINOR: remove unused scala files from core module (apache#9296) Reviewers: Mickael Maison <[email protected]>, Lee Dongjin <[email protected]> * MINOR: correct package of LinuxIoMetricsCollector (apache#9271) Reviewers: Mickael Maison <[email protected]>, Lee Dongjin <[email protected]> * KAFKA-10028: Minor fixes to describeFeatures and updateFeatures apis (apache#9393) In this PR, I have addressed the review comments from @chia7712 in apache#9001 which were provided after apache#9001 was merged. The changes are made mainly to KafkaAdminClient: Improve error message in updateFeatures api when feature name is empty. Propagate top-level error message in updateFeatures api. Add an empty-parameter variety for describeFeatures api. Minor documentation updates to @param and @return to make these resemble other apis. Reviewers: Chia-Ping Tsai [email protected], Jun Rao [email protected] * KAFKA-10271: Performance regression while fetching a key from a single partition (apache#9020) StreamThreadStateStoreProvider excessive loop over calling internalTopologyBuilder.topicGroups(), which is synchronized, thus causing significant performance degradation to the caller, especially when store has many partitions. Reviewers: John Roesler <[email protected]>, Guozhang Wang <[email protected]> Co-authored-by: Jorge Esteban Quilcate Otoya <[email protected]> Co-authored-by: Chia-Ping Tsai <[email protected]> Co-authored-by: Kowshik Prakasam <[email protected]> Co-authored-by: Dima Reznik <[email protected]>

…#11337) In #9139, we added backward iterator on SessionStore. But there is a bug that when fetch/backwardFetch the key range, if there are multiple records in the same session window, we can't return the data in the correct order. Reviewers: Anna Sophie Blee-Goldman <ableegoldman<apache.org>

…apache#11337) In apache#9139, we added backward iterator on SessionStore. But there is a bug that when fetch/backwardFetch the key range, if there are multiple records in the same session window, we can't return the data in the correct order. Reviewers: Anna Sophie Blee-Goldman <ableegoldman<apache.org>

jeqo mentioned this pull request Aug 7, 2020

KIP-617: Allow Kafka Streams State Stores to be iterated backwards #8976

Closed

3 tasks

jeqo force-pushed the backward-sessionstore branch 4 times, most recently from e7a774e to 07c31b9 Compare August 11, 2020 13:34

jeqo force-pushed the backward-sessionstore branch from 07c31b9 to 18fb88e Compare September 7, 2020 13:58

jeqo marked this pull request as ready for review September 7, 2020 14:27

jeqo force-pushed the backward-sessionstore branch from 3a045e5 to 5c5a04b Compare September 18, 2020 08:36

ableegoldman reviewed Oct 6, 2020

View reviewed changes

jeqo force-pushed the backward-sessionstore branch from 5c5a04b to 6909a44 Compare October 6, 2020 16:46

ableegoldman reviewed Oct 6, 2020

View reviewed changes

ableegoldman reviewed Oct 7, 2020

View reviewed changes

jeqo added 9 commits October 7, 2020 09:21

key/value reverse operation

9d4327c

window backward operations

6761c71

session backward operations

000c7b7

improve time range comments

1811b57

fix bytes range validator not needed

8c26756

fix cache iterator

e0d160b

fix session store to support backward queries

ae7e3ed

improve format based on feedback

64a95c0

test caching store across segments

a875cf1

jeqo force-pushed the backward-sessionstore branch from 3c8cd51 to a875cf1 Compare October 7, 2020 08:21

jeqo added 2 commits October 7, 2020 10:40

improve testing names and scenarios

76bb9a2

expand caching store tests per backing store

e49f258

ableegoldman approved these changes Oct 7, 2020

View reviewed changes

vvcephei approved these changes Oct 8, 2020

View reviewed changes

John Roesler added 2 commits October 7, 2020 22:36

fix formatting

b856242

fix formatting

1932413

John Roesler added 2 commits October 7, 2020 23:12

Merge branch 'trunk' into backward-sessionstore

70484e4

fix conflict with trunk

14fce52

vvcephei merged commit d0e6943 into apache:trunk Oct 8, 2020

vvcephei pushed a commit that referenced this pull request Oct 8, 2020

KAFKA-9929: Support backward iterator on SessionStore (#9139)

9390181

Implements KIP-617 for `SessionStore` Reviewers: A. Sophie Blee-Goldman <[email protected]>, John Roesler <[email protected]>

showuon mentioned this pull request Sep 18, 2021

KAFKA-13309: fix InMemorySessionStore#fetch/backwardFetch order issue #11337

Merged

3 tasks

jeqo deleted the backward-sessionstore branch October 19, 2021 14:21

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

KAFKA-9929: Support backward iterator on SessionStore #9139

KAFKA-9929: Support backward iterator on SessionStore #9139

jeqo commented Aug 7, 2020

jeqo commented Sep 11, 2020

ableegoldman left a comment

ableegoldman Oct 6, 2020

jeqo Oct 7, 2020

ableegoldman Oct 7, 2020

ableegoldman Oct 6, 2020

jeqo Oct 7, 2020

ableegoldman left a comment

ableegoldman Oct 7, 2020

jeqo Oct 7, 2020

ableegoldman commented Oct 7, 2020

ableegoldman left a comment

ableegoldman commented Oct 7, 2020

ableegoldman commented Oct 7, 2020

vvcephei left a comment

vvcephei Oct 7, 2020

vvcephei Oct 7, 2020

vvcephei Oct 7, 2020

vvcephei commented Oct 8, 2020

vvcephei commented Oct 8, 2020

vvcephei commented Oct 8, 2020

vvcephei commented Oct 8, 2020

vvcephei commented Oct 8, 2020

		public KeyValueIterator<Windowed<K>, V> backwardFetch(K from, K to) {
		return null;

	throw new UnsupportedOperationException("Moved from SessionStore");
	throw new UnsupportedOperationException("This API is not supported by this implementation of ReadOnlySessionStore.");

		if (forward) entries = nextKeyEntry.getValue().descendingMap().entrySet();
		else entries = nextKeyEntry.getValue().entrySet();

KAFKA-9929: Support backward iterator on SessionStore #9139

KAFKA-9929: Support backward iterator on SessionStore #9139

Conversation

jeqo commented Aug 7, 2020

Committer Checklist (excluded from commit message)

jeqo commented Sep 11, 2020

ableegoldman left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

ableegoldman left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

ableegoldman commented Oct 7, 2020

ableegoldman left a comment

Choose a reason for hiding this comment

ableegoldman commented Oct 7, 2020

ableegoldman commented Oct 7, 2020

vvcephei left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

vvcephei commented Oct 8, 2020

vvcephei commented Oct 8, 2020

vvcephei commented Oct 8, 2020

vvcephei commented Oct 8, 2020

vvcephei commented Oct 8, 2020