KAFKA-5172: Fix fetchPrevious to find the correct session. · Pull Request #2972 · apache/kafka

ghost · 2017-05-04T02:07:58Z

Change fetchPrevious to use findSessions with the proper key and timestamps rather than using fetch.

asfbot · 2017-05-04T03:16:23Z

Refer to this link for build results (access rights to CI server needed):
https://builds.apache.org/job/kafka-pr-jdk8-scala2.11/3463/
Test PASSed (JDK 8 and Scala 2.11).

asfbot · 2017-05-04T03:18:56Z

Refer to this link for build results (access rights to CI server needed):
https://builds.apache.org/job/kafka-pr-jdk7-scala2.10/3457/
Test PASSed (JDK 7 and Scala 2.10).

asfbot · 2017-05-04T03:46:57Z

Refer to this link for build results (access rights to CI server needed):
https://builds.apache.org/job/kafka-pr-jdk8-scala2.12/3454/
Test FAILed (JDK 8 and Scala 2.12).

guozhangwang · 2017-05-04T07:34:06Z

@dguy could you take a look?

dguy

@KyleWinkelman thanks for the PR. Can you please add a test to CachingSessionStoreTest that fails without this change?

ghost · 2017-05-04T12:34:47Z

@dguy I have added a test case that fails without this change.

asfbot · 2017-05-04T13:36:29Z

Refer to this link for build results (access rights to CI server needed):
https://builds.apache.org/job/kafka-pr-jdk7-scala2.10/3477/
Test FAILed (JDK 7 and Scala 2.10).

asfbot · 2017-05-04T13:56:47Z

Refer to this link for build results (access rights to CI server needed):
https://builds.apache.org/job/kafka-pr-jdk8-scala2.12/3474/
Test FAILed (JDK 8 and Scala 2.12).

asfbot · 2017-05-04T14:13:06Z

Refer to this link for build results (access rights to CI server needed):
https://builds.apache.org/job/kafka-pr-jdk8-scala2.11/3483/
Test FAILed (JDK 8 and Scala 2.11).

dguy

Thanks for adding the test @KyleWinkelman left a couple of comments

dguy · 2017-05-04T15:56:09Z

 */
 package org.apache.kafka.streams.state.internals;

+import static org.apache.kafka.streams.state.internals.RocksDBSessionStoreTest.toList;


can you re-order the input back to what they where before?

dguy · 2017-05-04T16:10:32Z

+    @Test
+    public void shouldPutAndMaybeForward() throws Exception {
+        final Windowed<String> a1 = new Windowed<>("a", new SessionWindow(0, 0));
+        final Windowed<String> a2 = new Windowed<>("a", new SessionWindow(0, 1));


I don't think we need to add the extra class just for this case. We could do something like:

final List<Change<Long>> values = new ArrayList<>(); cachingStore.setFlushListener(new CacheFlushListener<Windowed<String>, Long>() { @Override public void apply(final Windowed<String> key, final Long newValue, final Long oldValue) { values.add(new Change<>(newValue, oldValue)); } }); ... assertThat(values, equalTo(Arrays.asList(new Change<>(1L, null), new Change<>(null, 1L), new Change<>(2L, null))));

dguy · 2017-05-04T16:11:34Z

        assertFalse(results.hasNext());
    }

+    @Test


maybe something like: shouldForwardChangedValuesDuringFlush ?

dguy · 2017-05-04T16:11:52Z

+import org.apache.kafka.streams.kstream.internals.CacheFlushListener;
+import org.apache.kafka.streams.kstream.internals.Change;
+
+public class MockCacheFlushListener<K, V> implements CacheFlushListener<K, V> {


See comment above, i don't think we need this class

ghost · 2017-05-04T17:21:15Z

I made the recommended changes and altered the contents of the test a little so we could see a case where the new and old values are set.
Thanks for the comments.

asfbot · 2017-05-04T18:19:18Z

Refer to this link for build results (access rights to CI server needed):
https://builds.apache.org/job/kafka-pr-jdk8-scala2.11/3495/
Test PASSed (JDK 8 and Scala 2.11).

asfbot · 2017-05-04T18:28:20Z

Refer to this link for build results (access rights to CI server needed):
https://builds.apache.org/job/kafka-pr-jdk8-scala2.12/3486/
Test PASSed (JDK 8 and Scala 2.12).

asfbot · 2017-05-04T18:42:21Z

Refer to this link for build results (access rights to CI server needed):
https://builds.apache.org/job/kafka-pr-jdk7-scala2.10/3489/
Test FAILed (JDK 7 and Scala 2.10).

ghost · 2017-05-04T19:50:32Z

@dguy Do you think we should prevent the Bytes.wrap(serdes.rawKey(key.key())) from being called twice (once in putAndMaybeForward and once in fetchPrevious)? I don't know how costly this is but it could save a little bit of processing. I think we could call it once and change the fetchPrevious to accept Bytes and Window.

    private void putAndMaybeForward(final ThreadCache.DirtyEntry entry, final InternalProcessorContext context) {
        final Bytes binaryKey = entry.key();
        final RecordContext current = context.recordContext();
        context.setRecordContext(entry.recordContext());
        try {
            final Windowed<K> key = SessionKeySerde.from(binaryKey.get(), keySerde.deserializer(), topic);
            **final Bytes rawKey = Bytes.wrap(serdes.rawKey(key.key()));**
            if (flushListener != null) {
                final AGG newValue = serdes.valueFrom(entry.newValue());
                final AGG oldValue = fetchPrevious(rawKey, key.window());
                if (!(newValue == null && oldValue == null)) {
                    flushListener.apply(key, newValue, oldValue);
                }
            }
            bytesStore.put(new Windowed<>(**rawKey**, key.window()), entry.newValue());
        } finally {
            context.setRecordContext(current);
        }
    }

    private AGG fetchPrevious(**final Bytes rawKey, final Window window**) {
        try (final KeyValueIterator<Windowed<Bytes>, byte[]> iterator = bytesStore
                .findSessions(**rawKey**, window.start(), window.end())) {
            if (!iterator.hasNext()) {
                return null;
            }
            return serdes.valueFrom(iterator.next().value);
        }
    }

dguy · 2017-05-05T08:02:24Z

@KyleWinkelman yes that makes sense. Thanks!

ghost · 2017-05-05T13:12:51Z

Ok made the changes. Anything else required of me for this PR?

asfbot · 2017-05-05T14:12:45Z

Refer to this link for build results (access rights to CI server needed):
https://builds.apache.org/job/kafka-pr-jdk8-scala2.12/3537/
Test FAILed (JDK 8 and Scala 2.12).

dguy

Thanks @KyleWinkelman LGTM

asfbot · 2017-05-05T14:28:12Z

Refer to this link for build results (access rights to CI server needed):
https://builds.apache.org/job/kafka-pr-jdk8-scala2.11/3546/
Test FAILed (JDK 8 and Scala 2.11).

asfbot · 2017-05-05T14:37:39Z

Refer to this link for build results (access rights to CI server needed):
https://builds.apache.org/job/kafka-pr-jdk7-scala2.10/3540/
Test FAILed (JDK 7 and Scala 2.10).

guozhangwang · 2017-05-06T00:17:15Z

retest this please

asfbot · 2017-05-06T01:23:58Z

Refer to this link for build results (access rights to CI server needed):
https://builds.apache.org/job/kafka-pr-jdk8-scala2.11/3583/
Test PASSed (JDK 8 and Scala 2.11).

asfbot · 2017-05-06T01:42:51Z

Refer to this link for build results (access rights to CI server needed):
https://builds.apache.org/job/kafka-pr-jdk7-scala2.10/3577/
Test FAILed (JDK 7 and Scala 2.10).

asfbot · 2017-05-06T01:53:50Z

Refer to this link for build results (access rights to CI server needed):
https://builds.apache.org/job/kafka-pr-jdk8-scala2.12/3573/
Test FAILed (JDK 8 and Scala 2.12).

guozhangwang · 2017-05-07T18:38:06Z

retest this please

asfbot · 2017-05-07T19:44:32Z

Refer to this link for build results (access rights to CI server needed):
https://builds.apache.org/job/kafka-pr-jdk8-scala2.11/3620/
Test PASSed (JDK 8 and Scala 2.11).

asfbot · 2017-05-07T19:44:54Z

Refer to this link for build results (access rights to CI server needed):
https://builds.apache.org/job/kafka-pr-jdk7-scala2.10/3614/
Test FAILed (JDK 7 and Scala 2.10).

asfbot · 2017-05-07T20:21:03Z

Refer to this link for build results (access rights to CI server needed):
https://builds.apache.org/job/kafka-pr-jdk8-scala2.12/3611/
Test FAILed (JDK 8 and Scala 2.12).

guozhangwang

LGTM! Merged to trunk.

…for flushing / getter (#6161) #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).

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

Fix fetchPrevious to find the correct session.

c09a1ac

dguy reviewed May 4, 2017

View reviewed changes

Add test case to CachingSessionStoreTest for KAFKA-5172

a7acb05

ghost changed the title ~~[KAFKA-5172] Fix fetchPrevious to find the correct session.~~ KAFKA-5172: Fix fetchPrevious to find the correct session. May 4, 2017

dguy reviewed May 4, 2017

View reviewed changes

Recommended changes by reviewer.

b570979

Prevent duplicate serialization.

c8c8640

dguy approved these changes May 5, 2017

View reviewed changes

guozhangwang approved these changes May 8, 2017

View reviewed changes

asfgit closed this in e472ee7 May 8, 2017

ghost deleted the CachingSessionStore-fetchPrevious branch May 15, 2017 13:19

guozhangwang mentioned this pull request Jan 17, 2019

KAFKA-7652: Part II; Add single-point query for SessionStore and use for flushing / getter #6161

Merged

3 tasks

Conversation

ghost commented May 4, 2017

Uh oh!

asfbot commented May 4, 2017

Uh oh!

asfbot commented May 4, 2017

Uh oh!

asfbot commented May 4, 2017

Uh oh!

guozhangwang commented May 4, 2017

Uh oh!

dguy left a comment

Choose a reason for hiding this comment

Uh oh!

ghost commented May 4, 2017

Uh oh!

asfbot commented May 4, 2017

Uh oh!

asfbot commented May 4, 2017

Uh oh!

asfbot commented May 4, 2017

Uh oh!

dguy left a comment

Choose a reason for hiding this comment

Uh oh!

dguy May 4, 2017

Choose a reason for hiding this comment

Uh oh!

dguy May 4, 2017

Choose a reason for hiding this comment

Uh oh!

dguy May 4, 2017

Choose a reason for hiding this comment

Uh oh!

dguy May 4, 2017

Choose a reason for hiding this comment

Uh oh!

ghost commented May 4, 2017

Uh oh!

asfbot commented May 4, 2017

Uh oh!

asfbot commented May 4, 2017

Uh oh!

asfbot commented May 4, 2017

Uh oh!

ghost commented May 4, 2017 • edited by ghost Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

dguy commented May 5, 2017

Uh oh!

ghost commented May 5, 2017

Uh oh!

asfbot commented May 5, 2017

Uh oh!

dguy left a comment

Choose a reason for hiding this comment

Uh oh!

asfbot commented May 5, 2017

Uh oh!

asfbot commented May 5, 2017

Uh oh!

guozhangwang commented May 6, 2017

Uh oh!

asfbot commented May 6, 2017

Uh oh!

asfbot commented May 6, 2017

Uh oh!

asfbot commented May 6, 2017

Uh oh!

guozhangwang commented May 7, 2017

Uh oh!

asfbot commented May 7, 2017

Uh oh!

asfbot commented May 7, 2017

Uh oh!

asfbot commented May 7, 2017

Uh oh!

guozhangwang left a comment

Choose a reason for hiding this comment

Uh oh!

ghost commented May 4, 2017 •

edited by ghost

Loading