KAFKA-13309: fix InMemorySessionStore#fetch/backwardFetch order issue by showuon · Pull Request #11337 · apache/kafka

showuon · 2021-09-18T02:46:30Z

JIRA is here.
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.

For example:
We have a session window inactivity gap with 10 ms, and the records:

key: "A", value: "AA", timestamp: 0 --> with SessionWindow(0, 0)
key: "B", value: "BB", timestamp: 0 --> with SessionWindow(0, 0)
key: "C", value: "CC", timestamp: 0 --> with SessionWindow(0, 0)
key: "D" value: "DD", timestamp: 100 --> with SessionWindow(100, 100)

So, when fetch("A" /key from/, "D" /key to/), we expected to have [A, B, C, D], but we'll have [C, B A, D ]

And the reason is that we pass "false" in the "is forward" parameter for fetch method, and "true" for "backwardFetch" method, which obviously is wrong.

So, why does the tests can't find this issue?
It's because the test data we provided doesn't have multiple data in the same session window.

In this PR, I fixed the issue, and add tests to improve the test coverage.

Committer Checklist (excluded from commit message)

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

showuon · 2021-09-18T02:58:33Z



-        return registerNewIterator(keyFrom, keyTo, Long.MAX_VALUE, endTimeMap.entrySet().iterator(), false);
+        return registerNewIterator(keyFrom, keyTo, Long.MAX_VALUE, endTimeMap.entrySet().iterator(), true);


fix: to make fetch method passing true for isFarwarded.

showuon · 2021-09-18T02:58:47Z


        return registerNewIterator(
-            keyFrom, keyTo, Long.MAX_VALUE, endTimeMap.descendingMap().entrySet().iterator(), true);
+            keyFrom, keyTo, Long.MAX_VALUE, endTimeMap.descendingMap().entrySet().iterator(), false);


fix: to make backwardFetch method passing false for isFarwarded.

showuon · 2021-09-18T03:01:36Z

        try (final KeyValueIterator<Windowed<String>, Long> values = sessionStore.findSessions(key, 0, 1000L)
        ) {
-            assertEquals(new HashSet<>(expected), toSet(values));
+            assertEquals(expected, toList(values));


before this PR, we only verify the returned data contains in the expected data as Set. Now, we'll change to List, which means, not only the data should be correct, but also the data order should be correct.

showuon · 2021-09-18T03:04:01Z

+        final List<KeyValue<Windowed<String>, Long>> expected = new LinkedList<>();
+        expected.add(KeyValue.pair(new Windowed<>("aa", new SessionWindow(10, 10)), 2L));
+        expected.add(KeyValue.pair(new Windowed<>("aaa", new SessionWindow(100, 100)), 3L));
+        expected.add(KeyValue.pair(new Windowed<>("aaaa", new SessionWindow(100, 100)), 6L));


add a record with the same SessionWindow(100, 100). So we can test if the fetch/backwardFetch can return data in correct order for both InMemorySessionStore and RocksDBSessionStore.

showuon · 2021-09-18T03:05:35Z

-        final Set<String> expectedKey1 = new HashSet<>(asList("1", "4", "7"));
-        assertThat(valuesToSet(sessionStore.findSessions(key1, 0L, Long.MAX_VALUE)), equalTo(expectedKey1));
-        final Set<String> expectedKey2 = new HashSet<>(asList("2", "5", "8"));
-        assertThat(valuesToSet(sessionStore.findSessions(key2, 0L, Long.MAX_VALUE)), equalTo(expectedKey2));
-        final Set<String> expectedKey3 = new HashSet<>(asList("3", "6", "9"));
-        assertThat(valuesToSet(sessionStore.findSessions(key3, 0L, Long.MAX_VALUE)), equalTo(expectedKey3));
+        final List<String> expectedKey1 = asList("1", "4", "7");
+        try (KeyValueIterator<Windowed<Bytes>, String> iterator = sessionStore.findSessions(key1, 0L, Long.MAX_VALUE)) {
+            assertThat(valuesToSet(iterator), equalTo(new HashSet<>(expectedKey1)));
+        }


We forgot to close the KeyValueIterator instance returned by the findSessions. Fix it.

showuon · 2021-09-18T03:07:19Z

@jeqo @ableegoldman @vvcephei , please help review this PR. Thank you.

showuon · 2021-09-27T07:18:58Z

@jeqo @ableegoldman @vvcephei @guozhangwang , please help review this PR when available. Thank you.

ableegoldman

Thanks for the patch! LGTM

…#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>

ableegoldman · 2021-09-29T00:54:02Z

Merged to trunk and cherrypicked to 3.0

) When cherrypicking #11337 back to the 3.0 branch and resolving a minor merge conflict, I forgot to git add the removal of some imports that were breaking checkstyle. Reviewers: Natea Eshetu Beshada <nbeshada@confluent.io>, Walker Carlson <wcarlson@confluent.io>

…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>

…che#11365) When cherrypicking apache#11337 back to the 3.0 branch and resolving a minor merge conflict, I forgot to git add the removal of some imports that were breaking checkstyle. Reviewers: Natea Eshetu Beshada <nbeshada@confluent.io>, Walker Carlson <wcarlson@confluent.io>

showuon added 3 commits August 19, 2021 14:50

KAFKA-13212: add support infinite query for session store

45db1f6

Merge branch 'trunk' of https://github.com/apache/kafka into KAFKA-13212

39de0d0

revert

c9ef98d

showuon force-pushed the KAFKA-13309 branch from dd5fd06 to c9ef98d Compare September 18, 2021 02:57

showuon commented Sep 18, 2021

View reviewed changes

ableegoldman approved these changes Sep 29, 2021

View reviewed changes

ableegoldman merged commit 361b784 into apache:trunk Sep 29, 2021

ableegoldman mentioned this pull request Sep 29, 2021

HOTFIX: fix checkstyle after cherrypick #11365

Merged

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

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

KAFKA-13309: fix InMemorySessionStore#fetch/backwardFetch order issue#11337
ableegoldman merged 3 commits intoapache:trunkfrom
showuon:KAFKA-13309

showuon commented Sep 18, 2021

Uh oh!

showuon Sep 18, 2021

Uh oh!

showuon Sep 18, 2021

Uh oh!

showuon Sep 18, 2021

Uh oh!

showuon Sep 18, 2021 •

edited

Loading

Uh oh!

showuon Sep 18, 2021

Uh oh!

showuon commented Sep 18, 2021

Uh oh!

showuon commented Sep 27, 2021

Uh oh!

ableegoldman left a comment

Uh oh!

ableegoldman commented Sep 29, 2021

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

2 participants



		return registerNewIterator(keyFrom, keyTo, Long.MAX_VALUE, endTimeMap.entrySet().iterator(), false);
		return registerNewIterator(keyFrom, keyTo, Long.MAX_VALUE, endTimeMap.entrySet().iterator(), true);

Conversation

showuon commented Sep 18, 2021

Committer Checklist (excluded from commit message)

Uh oh!

showuon Sep 18, 2021

Choose a reason for hiding this comment

Uh oh!

showuon Sep 18, 2021

Choose a reason for hiding this comment

Uh oh!

showuon Sep 18, 2021

Choose a reason for hiding this comment

Uh oh!

showuon Sep 18, 2021 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

showuon Sep 18, 2021

Choose a reason for hiding this comment

Uh oh!

showuon commented Sep 18, 2021

Uh oh!

showuon commented Sep 27, 2021

Uh oh!

ableegoldman left a comment

Choose a reason for hiding this comment

Uh oh!

ableegoldman commented Sep 29, 2021

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

2 participants

showuon Sep 18, 2021 •

edited

Loading