KAFKA-7652: Part II; Add single-point query for SessionStore and use for flushing / getter by guozhangwang · Pull Request #6161 · apache/kafka

guozhangwang · 2019-01-17T07:39:53Z

#2972 tried to fix a bug about flushing operation, but it was not complete, since findSessions(key, earliestEnd, latestStart) does not guarantee to only return a single entry since its semantics are to return any sessions whose end > earliestEnd and whose start < latestStart.

I've tried various ways to fix it completely and I ended up having to add a single-point query to the public ReadOnlySessionStore API for the exact needed semantics. It is used for flushing to read the old values (otherwise the wrong old values will be sent downstreams, hence it is a correctness issue) and also for getting the value for value-getters (it is for perf only).

Committer Checklist (excluded from commit message)

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

…derbytes-upper-range

guozhangwang · 2019-01-17T07:45:35Z

                final AGG newValue = serdes.valueFrom(entry.newValue());
-                final AGG oldValue = newValue == null || sendOldValues ? fetchPrevious(rawKey, key.window()) : null;
+                final AGG oldValue = newValue == null || sendOldValues ?
+                    serdes.valueFrom(bytesStore.fetch(rawKey, key.window().start(), key.window().end())) :


This is the actual fix.

guozhangwang · 2019-01-17T07:45:47Z

-                }
-                return value;
-            }
+            return store.fetch(key.key(), key.window().start(), key.window().end());


This is a good-to-have for perf.

guozhangwang · 2019-01-17T07:46:41Z

+        final Windowed<Bytes> a = new Windowed<>(keyA, new SessionWindow(2, 4));
+        final Windowed<Bytes> b = new Windowed<>(keyA, new SessionWindow(1, 2));
+        final Windowed<String> aDeserialized = new Windowed<>("a", new SessionWindow(2, 4));
+        final Windowed<String> bDeserialized = new Windowed<>("a", new SessionWindow(1, 2));


This additional record is for illustrating the bug. Without the fix the flushed entries would be incorrect.

guozhangwang · 2019-01-17T07:47:33Z


-    public static byte[] toBinary(final Windowed<Bytes> sessionKey) {
-        final byte[] bytes = sessionKey.key().get();
+    public static Bytes toBinary(final Bytes key,


This is just some refactoring of the code to 1) merge the logic into a single function and 2) do Bytes.wrap internally so that callers do not need to call.

nice addition!

bbejeck

Thanks for the patch @guozhangwang. I made a pass over this, and I have a couple of minor comments, but otherwise, this looks good to me.

I do think we need a KIP for the addition to the ReadonlySessionStore, but the use case clean and straightforward so IMHO we should be able to get this in for 2.2.

bbejeck · 2019-01-17T18:53:27Z

-                }
+        cache.addDirtyEntryFlushListener(cacheName, entries -> {
+            for (final ThreadCache.DirtyEntry entry : entries) {
+                putAndMaybeForward(entry, context);


nice cleanup!

bbejeck · 2019-01-17T19:00:14Z

+        validateStoreOpen();
+        final Bytes bytesKey = SessionKeySchema.toBinary(key, startTime, endTime);
+        final Bytes cacheKey = cacheFunction.cacheKey(bytesKey);
+        if (cache == null) {


nit: move the cache == null check two lines up and save creating the bytesKey and cacheKey operations when the cache is null. I know this is a minor point, but IMHO still worth the change.

bbejeck · 2019-01-17T19:08:00Z

+            } catch (final InvalidStateStoreException ise) {
+                throw new InvalidStateStoreException("State store  [" + storeName + "] is not available anymore" +
+                    " and may have been migrated to another instance; " +
+                    "please re-discover its location from the state metadata.");


nit: do we want to include the original InvalidStateStoreException (ise) in the new exception we are throwing?

We intentionally change the error message with the same InvalidStateStoreException type in the original design. But thinking about it again, I think it still worth exposing the original exception for debugging purposes.

bbejeck · 2019-01-17T19:11:46Z


+    @Override
+    public AGG fetch(final K key, final long startTime, final long endTime) {
+        return serdes.valueFrom(bytesStore.get(SessionKeySchema.toBinary(Bytes.wrap(serdes.rawKey(key)), startTime, endTime)));


nit: validate key is not null here?

This is validated in higher in the call trace (MeteredXX) already.

bbejeck · 2019-01-17T19:17:15Z


-    public static byte[] toBinary(final Windowed<Bytes> sessionKey) {
-        final byte[] bytes = sessionKey.key().get();
+    public static Bytes toBinary(final Bytes key,


nice addition!

…derbytes-upper-range

guozhangwang · 2019-01-20T21:55:34Z

@bbejeck updated per comments.

bbejeck

Thanks @guozhangwang LGTM

…derbytes-upper-range

vvcephei

LGTM. Thanks!

vvcephei · 2019-01-29T06:53:50Z

                                                             " and may have been migrated to another instance; " +
-                                                             "please re-discover its location from the state metadata.");
+                                                             "please re-discover its location from the state metadata. " +
+                                                             "Original error message: " + ise.toString());


Why not just include ise as the cause? Do we want to hide its stacktrace of something?

Yes this is intentional indeed. Since this is a user-facing interface we were abstracting the internal stacktrace from users.

mjsax

LGTM.

…derbytes-upper-range

guozhangwang · 2019-01-31T01:31:17Z

KIP-420 (https://cwiki.apache.org/confluence/display/KAFKA/KIP-420%3A+Add+Single+Value+Fetch+in+Session+Stores) has been voted before the 2.2 KIP deadline. Merging to trunk now.

* AK/trunk: fix typo (apache#5150) MINOR: Reduce replica.fetch.backoff.ms in ReassignPartitionsClusterTest (apache#5887) KAFKA-7766: Fail fast PR builds (apache#6059) KAFKA-7798: Expose embedded clientIds (apache#6107) KAFKA-7641; Introduce "group.max.size" config to limit group sizes (apache#6163) KAFKA-7433; Introduce broker options in TopicCommand to use AdminClient (KIP-377) MINOR: Fix some field definitions for ListOffsetReponse (apache#6214) KAFKA-7873; Always seek to beginning in KafkaBasedLog (apache#6203) KAFKA-7719: Improve fairness in SocketServer processors (KIP-402) (apache#6022) MINOR: fix checkstyle suppressions for generated RPC code to work on Windows KAFKA-7859: Use automatic RPC generation in LeaveGroups (apache#6188) KAFKA-7652: Part II; Add single-point query for SessionStore and use for flushing / getter (apache#6161) KAFKA-3522: Add RocksDBTimestampedStore (apache#6149) KAFKA-3522: Replace RecordConverter with TimestampedBytesStore (apache#6204)

…for flushing / getter (apache#6161) apache#2972 tried to fix a bug about flushing operation, but it was not complete, since findSessions(key, earliestEnd, latestStart) does not guarantee to only return a single entry since its semantics are to return any sessions whose end > earliestEnd and whose start < latestStart. I've tried various ways to fix it completely and I ended up having to add a single-point query to the public ReadOnlySessionStore API for the exact needed semantics. It is used for flushing to read the old values (otherwise the wrong old values will be sent downstreams, hence it is a correctness issue) and also for getting the value for value-getters (it is for perf only).

guozhangwang added 5 commits January 15, 2019 16:09

first try

2723846

add single point query

f08e0a5

Merge branch 'trunk' of https://github.com/apache/kafka into K7652-or…

152f017

…derbytes-upper-range

unit tests

c739655

checkstyle fixes

d0a12ea

guozhangwang commented Jan 17, 2019

View reviewed changes

bbejeck reviewed Jan 17, 2019

View reviewed changes

guozhangwang added 7 commits January 18, 2019 09:55

Merge branch 'trunk' of https://github.com/apache/kafka into K7652-or…

bddeae0

…derbytes-upper-range

move to SessionStore

a51380b

one minor fix

5c6cab3

rebase from trunk

ea7b4f1

fix unit tests

862f79f

Merge branch 'trunk' of https://github.com/apache/kafka into K7652-or…

6f2ec56

…derbytes-upper-range

github comments

ec3ec15

bbejeck approved these changes Jan 21, 2019

View reviewed changes

Merge branch 'trunk' of https://github.com/apache/kafka into K7652-or…

3a25ef7

…derbytes-upper-range

mjsax added the streams label Jan 27, 2019

vvcephei approved these changes Jan 29, 2019

View reviewed changes

mjsax approved these changes Jan 30, 2019

View reviewed changes

Merge branch 'trunk' of https://github.com/apache/kafka into K7652-or…

f4bcfd8

…derbytes-upper-range

guozhangwang merged commit 9d5893d into apache:trunk Jan 31, 2019

Conversation

guozhangwang commented Jan 17, 2019

Committer Checklist (excluded from commit message)

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

bbejeck left a comment • 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!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

guozhangwang commented Jan 20, 2019

Uh oh!

bbejeck left a comment

Choose a reason for hiding this comment

Uh oh!

vvcephei left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

mjsax left a comment

Choose a reason for hiding this comment

Uh oh!

guozhangwang commented Jan 31, 2019

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

4 participants

bbejeck left a comment •

edited

Loading