KAFKA-3452: Support session windows by dguy · Pull Request #2166 · apache/kafka

dguy · 2016-11-24T16:48:23Z

Add support for SessionWindows based on design detailed in https://cwiki.apache.org/confluence/display/KAFKA/KIP-94+Session+Windows.
This includes refactoring of the RocksDBWindowStore such that functionality common with the RocksDBSessionStore isn't duplicated.

dguy · 2016-11-24T16:54:26Z

@enothereska @mjsax @guozhangwang.

This follows the design as proposed in the KIP.
Some key things:
I've extracted all of the Segment functionality out of RocksDBWindowStore and into Segments. I've introduced a new class SegmentedBytesStore that both the RocksDBWindowStore and RocksDBSessionStore delegate to. They are wrappers on top of the SegmentedBytesStore that know how to deal with the specifics to those stores, i.e., iterators, key layout etc.
The new class SessionWindows doesn't derive from Windows as it doesn't make sense to pass a SessionWindows instance to the non-sesssion-window aggregate methods on KGroupedStream

Thanks

enothereska

Initial pass on just high level APIs and high-level implementation only.

enothereska · 2016-11-28T14:04:54Z

+    /**
+     * Combine values of this stream by key into {@link SessionWindows}
+     * The resulting {@link KTable} will be materialized in a local state
+     * store with the given store name. Also a changelog topic named "${applicationId}-${storeName}-changelog"


StoreName is not defined in this API.

enothereska · 2016-11-28T14:05:17Z

+
+    /**
+     * Aggregate values of this stream by key into {@link SessionWindows}.
+     * The resulting {@link KTable} will be materialized in a local state


Same here with storename.

enothereska · 2016-11-28T14:06:05Z

+    /**
+     * Count number of records of this stream by key into {@link SessionWindows}.
+     * The resulting {@link KTable} will be materialized in a local state
+     * store with the given store name. Also a changelog topic named "${applicationId}-${storeName}-changelog"


Same here with storename.

enothereska · 2016-11-28T14:07:31Z

+ * +-----------+-------------+------------+
+ *
+ * The previous 2 sessions would be merged into a single session with start time 10 and end time 20.
+ * The aggregate value for this session would be the result of aggregating all 4 values.


Nice explanation!

enothereska · 2016-11-28T14:14:16Z

+import org.apache.kafka.streams.kstream.Windowed;
+
+import java.nio.ByteBuffer;
+


Consider adding a line or two on what this class is meant for.

This is a general thing we lack for many internal classes... It's hard to get started if it's not clear what a class does and even more important why.

The comment added is actually quite good!

enothereska · 2016-11-28T14:17:26Z

@@ -25,21 +25,24 @@
 class TupleForwarder<K, V> {


Can we add a line to describe what this class is for?

enothereska · 2016-11-28T14:20:28Z

                8 + // offset
                4 + // partition
-                topic.length();
+                (topic == null ? 0 : topic.length());


Good catch!

enothereska · 2016-11-28T14:22:06Z

            }
            entries.add(new ThreadCache.DirtyEntry(key, node.entry.value, node.entry));
            node.entry.markClean();
+            if (node.entry.value == null) {


So if value==null we automatically delete?

Yes - automatically delete from the cache.

mjsax · 2016-11-29T19:36:14Z

+     * @param aggTwo    the second aggregate
+     * @return          the new aggregate value
+     */
+    V apply(K aggKey, V aggOne, V aggTwo);


Do we need the key?

It is consistent with Aggregator - also, why not? Maybe a developer wants to do something with the key.

It's a matter of taste I guess. For example ValueJoiner has no key argument either (there was a discussion about adding the key though...). The API is nor really consistent right now. Should we try to make it consisted adding key parameter everywhere?
\cc @enothereska @miguno @guozhangwang

mjsax · 2016-11-29T19:36:50Z

+
+
+/**
+ * A session based window specification uses for aggregating events into sessions.


nit: uses -> used

mjsax · 2016-11-29T19:51:49Z

+ * then a new session will be created.
+ *
+ * For example, If we have a session gap of 5 and the following data arrives:
+ * +--------------------------------------+


nit: this will not render nicely as compiled JavaDoc. Can you add some markup?

mjsax · 2016-11-29T20:36:16Z

+     * @return a new SessionWindows with the provided inactivity gap
+     * and default maintain duration
+     */
+    public static SessionWindows inactivityGap(final long gapMs) {


What about with(final long inactivityGap) instead of inactivityGap(final long gapMs)? This would be closer to TimeWindows#of(long size) ?

mjsax · 2016-11-29T22:16:24Z

+        final SessionMerger<K, V> sessionMerger = new SessionMerger<K, V>() {
+            @Override
+            public V apply(final K aggKey, final V aggOne, final V aggTwo) {
+                return aggregator.apply(aggKey, aggTwo, aggOne);


We have to document that the aggregator is used a SessionMerger.

mjsax · 2016-11-29T23:25:59Z

 * @see org.apache.kafka.streams.state.Stores#create(String)
 */
-public class RocksDBStore<K, V> implements KeyValueStore<K, V> {
+class RocksDBStore<K, V> implements KeyValueStore<K, V> {


Why not allow developers to use it?

It is in the internals package and i believe in making internal code as private as possible. If they want to use the RocksDBStore they can already create one via the various suppliers etc.

All that said, though, this could break anyone that has sub-classed this. So i'll change it back.

mjsax · 2016-11-29T23:40:28Z

+                        new Properties()),
+                t1);
+
+        final long t2 = t1 + (sessionGap / 2);


final long t2 = t1 + sessionGap;
to test the corner case

nah - that is what i wanted. There are further tests in KStreamSessionWindowAggregateProcessorTest

mjsax · 2016-11-29T23:46:31Z

+
+    }
+



Could you add a test, that does merge two session into a single one, because a late record arrives (ie late record is in between both session) ?

Is already covered in KStreamSessionWindowAggregateProcessorTest

mjsax · 2016-11-30T00:08:09Z

+                        StringSerializer.class,
+                        new Properties()),
+                t2);
+        final long t3 = t1 + sessionGap;


final long t3 = t1 + sessionGap + 1; -- otherwise I would assume that the sessions get merged.

mjsax · 2016-11-30T00:12:33Z

    @Override
    public void schedule(long interval) {
-        throw new UnsupportedOperationException("schedule() not supported.");
+//        throw new UnsupportedOperationException("schedule() not supported.");


Good question!

asfbot · 2016-12-12T16:21:12Z

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

asfbot · 2016-12-12T16:21:14Z

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

asfbot · 2016-12-12T16:21:45Z

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

enothereska · 2016-12-14T13:54:05Z

+                                  final String storeName);
+
+    /**
+     * Combine values of this stream by the grouped key into {@link SessionWindows}.


Do we need repeated Javadoc everywhere? Not sure what's best practices here. Just there is a lot of repetition and perhaps one aggregate can have the Javadoc and others can point to it?

I know - it is consistent with the rest of the class.

Agree that it is a problem, but if you use an IDE and hoover over a class, it should display it -- and not just a link to another method... I would be awesome if JavaDoc would allow to factor common stuff out for overloaded methods. But I don't know a good way to do it either -- there is only c&p... :(

enothereska · 2016-12-14T13:58:26Z

+     *
+     * @return  itself
+     */
+    public SessionWindows until(long durationMs) {


Should duration be final?

This one yes.

enothereska · 2016-12-14T13:59:52Z

+     * Fetch any session aggregates with the matching key and the sessions end is >= earliestEndTime and the sessions
+     * start is <= latestStartTime
+     */
+    KeyValueIterator<Windowed<K>, AGG> findSessionsToMerge(K key, long earliestSessionEndTime, long latestSessionStartTime);


Should parameters be final for consistency with your other code?

enothereska · 2016-12-14T14:01:57Z

+     *         where each table contains records with unmodified keys and values
+     *         that represent the latest (rolling) count (i.e., number of records) for each key within that window
+     */
+    KTable<Windowed<K>, Long> count(final SessionWindows sessionWindows, final String storeName);


This probably doesn't belong to this PR, but I wonder why we need count at all, it's just an aggregate.

I guess, it's THE aggregate everyone wants. Syntactic sugar.

enothereska · 2016-12-14T14:28:32Z

+                while (iterator.hasNext()) {
+                    final KeyValue<Windowed<K>, T> next = iterator.next();
+                    merged.add(next);
+                    agg = sessionMerger.apply(key, agg, next.value);


This would have probably better been part of the KIP discussion, but there will be some aggregates, like Average for which this kind of merging won't work because they will need more than just the previous aggregate values. How do we catch such cases and warn the user?

This is the same for non SessionWindows based aggregates. I.e, if you aggregate an int into an Average you cant just apply the new int to the currently aggregated Average.

enothereska · 2016-12-14T14:31:11Z

+            if (!mergedWindow.equals(newTimeWindow)) {
+                for (final KeyValue<Windowed<K>, T> session : merged) {
+                    store.remove(session.key);
+                    tupleForwarder.maybeForward(session.key, null, session.value);


Don't we want to first invalidate previous values, then send new values, i.e., flip the order of this forwarding with the previous one? I can't immediately find the Javadoc where we tell the user the order.

enothereska · 2016-12-14T14:44:55Z

+        };
+    }
+
+    static class SessionStoreIterator<K, AGG> implements KeyValueIterator<Windowed<K>, AGG> {


Why is this part of RocksDbSessionStore? Isn't it useful for stores other than RocksDb-based?

It doesn't strictly need to be part of RocksDbSessionStore, but it is only used here as of now. If it is needed elsewhere later, then we should move it later.

enothereska · 2016-12-14T14:45:51Z

+class CachingSessionStore<K, AGG>  implements SessionStore<K, AGG>, CachedStateStore<Windowed<K>, AGG> {
+
+    private final SegmentedBytesStore bytesStore;
+    private final RocksDBSessionStore.SessionKeySchema keySchema;


If we use a different store, not RocksDb, will this variable still be good?

Yes - it applies to a SegmentedBytesStore - which is not RocksDB specific. The class could be moved out of RocksDBSessionStore, but I don't really see the need to at the moment.

enothereska · 2016-12-14T14:52:32Z

+            public void apply(final List<ThreadCache.DirtyEntry> entries) {
+                for (ThreadCache.DirtyEntry entry : entries) {
+                    final Bytes binaryKey = entry.key();
+                    final RecordContext current = context.recordContext();


In previous stores, like CachingKeyValueStore, you put the code below in a separate method called putAndMaybeForward that I thought was a neat way to break the code into chunks.

enothereska · 2016-12-14T14:53:19Z

+    }
+
+    public void close() {
+        bytesStore.close();


Do we want to flush and close the cache too?

Yeah. The closing of the cache was dependent on another PR that has now been merged. flush doesn't matter so much as we always flush and then close, but i'll add it anyway

asfbot · 2016-12-14T14:54:13Z

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

asfbot · 2016-12-14T14:55:49Z

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

asfbot · 2016-12-14T15:27:52Z

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

asfbot · 2016-12-14T15:58:20Z

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

enothereska · 2016-12-14T15:51:43Z

+        Objects.requireNonNull(sessionMerger, "sessionMerger can't be null");
+        Objects.requireNonNull(storeSupplier, "storeSupplier can't be null");
+
+        return (KTable<Windowed<K>, T>) doAggregate(


Is the casting needed?

It is consistent with what is already there. Also, i can't think of another way of doing it without duplicating doAggregate, which i'd rather not do

asfbot · 2016-12-14T16:02:31Z

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

enothereska · 2016-12-14T16:02:51Z

@dguy I had a look over:

APIs
main algo for merging sessions
session store and IQ read only session store
kgroupedstreamimpl
basic skimming over new tests (they look great!).

Apart from the above comments I don't have anything else. Thanks.

asfbot · 2016-12-14T16:10:26Z

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

dguy · 2017-01-05T12:33:56Z

@guozhangwang i've addressed your comments.
w.r.t your suggested refactoring: though i agree the hierarchy could do with some tidying up, i'm not 100% sure what you are suggesting will work and a lot of it is un-related to this PR. I think it needs some more thought and this PR has been outstanding for nearly 2 months already. Also, the changes you are suggesting are internal (i.e., not affecting public APIs), so they could be done once this PR is merged. I think we might already have a JIRA for creating Logged**Stores ? Will have to look.

asfbot · 2017-01-05T12:47:35Z

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

asfbot · 2017-01-05T13:49:30Z

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

asfbot · 2017-01-05T14:03:06Z

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

asfbot · 2017-01-05T15:31:40Z

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

asfbot · 2017-01-05T15:34:22Z

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

asfbot · 2017-01-05T15:41:29Z

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

guozhangwang · 2017-01-05T17:30:04Z

I think it needs some more thought and this PR has been outstanding for nearly 2 months already.

I understand... Just afraid such tech debt will keep growing and eventually come back and bite us as unstable and hard-to-debug code. Well how about merging this PR as is and see if we can still push it as a separate PR before the release deadline?

dguy · 2017-01-06T09:18:28Z

@guozhangwang - that is fine with me. I agree with the tech-debt and i think we should really focus the next release on bug fixing, tech-debt, usability etc.
If we get this in i'm more than happy to tackle your suggestions above

guozhangwang · 2017-01-06T18:11:17Z

@dguy Cool, will merge it as is. Note that I have a couple follow-up comments in the previous round, but I think it would be easier to address them as a new PR than this 5000+ LOC one.

This is a follow up of #2166 - refactoring the store hierarchies as requested Author: Damian Guy <damian.guy@gmail.com> Reviewers: Guozhang Wang <wangguoz@gmail.com> Closes #2360 from dguy/state-store-refactor (cherry picked from commit 73b7ae0) Signed-off-by: Guozhang Wang <wangguoz@gmail.com>

This is a follow up of #2166 - refactoring the store hierarchies as requested Author: Damian Guy <damian.guy@gmail.com> Reviewers: Guozhang Wang <wangguoz@gmail.com> Closes #2360 from dguy/state-store-refactor

This is a refactoring follow-up of #2166. Main refactoring changes: 1. Extract `InMemoryKeyValueStore` out of `InMemoryKeyValueStoreSupplier` and remove its duplicates in test package. 2. Add two abstract classes `AbstractKeyValueIterator` and `AbstractKeyValueStore` to collapse common functional logics. 3. Added specialized `BytesXXStore` to accommodate cases where key value types are Bytes / byte[] so that we can save calling the dummy serdes. 4. Make the key type in `ThreadCache` from byte[] to Bytes, as SessionStore / WindowStore's result serialized bytes are in the form of Bytes anyways, so that we can save unnecessary `Bytes.get()` and `Bytes.wrap(bytes)`. Each of these should arguably be a separate PR and I apologize for the mess, this is because this branch was extracted from a rather large diff that has multiple refactoring mingled together and dguy and myself have already put lots of efforts to break it down to a few separate PRs, and this is the only left-over work. Such PR won't happen in the future. Ping dguy enothereska mjsax for reviews Author: Guozhang Wang <wangguoz@gmail.com> Reviewers: Damian Guy, Matthias J. Sax, Jun Rao Closes #2333 from guozhangwang/K3452-followup-state-store-refactor

Add support for SessionWindows based on design detailed in https://cwiki.apache.org/confluence/display/KAFKA/KIP-94+Session+Windows. This includes refactoring of the RocksDBWindowStore such that functionality common with the RocksDBSessionStore isn't duplicated. Author: Damian Guy <damian.guy@gmail.com> Reviewers: Eno Thereska <eno.thereska@gmail.com>, Matthias J. Sax <matthias@confluent.io>, Guozhang Wang <wangguoz@gmail.com> Closes apache#2166 from dguy/kafka-3452-session-merge

This is a follow up of apache#2166 - refactoring the store hierarchies as requested Author: Damian Guy <damian.guy@gmail.com> Reviewers: Guozhang Wang <wangguoz@gmail.com> Closes apache#2360 from dguy/state-store-refactor

This is a refactoring follow-up of apache#2166. Main refactoring changes: 1. Extract `InMemoryKeyValueStore` out of `InMemoryKeyValueStoreSupplier` and remove its duplicates in test package. 2. Add two abstract classes `AbstractKeyValueIterator` and `AbstractKeyValueStore` to collapse common functional logics. 3. Added specialized `BytesXXStore` to accommodate cases where key value types are Bytes / byte[] so that we can save calling the dummy serdes. 4. Make the key type in `ThreadCache` from byte[] to Bytes, as SessionStore / WindowStore's result serialized bytes are in the form of Bytes anyways, so that we can save unnecessary `Bytes.get()` and `Bytes.wrap(bytes)`. Each of these should arguably be a separate PR and I apologize for the mess, this is because this branch was extracted from a rather large diff that has multiple refactoring mingled together and dguy and myself have already put lots of efforts to break it down to a few separate PRs, and this is the only left-over work. Such PR won't happen in the future. Ping dguy enothereska mjsax for reviews Author: Guozhang Wang <wangguoz@gmail.com> Reviewers: Damian Guy, Matthias J. Sax, Jun Rao Closes apache#2333 from guozhangwang/K3452-followup-state-store-refactor

MarcoAbi · 2017-02-21T21:50:10Z

Hello,

i'm having an issue trying to use SessionWindows

This exception is thrown:

Exception in thread "StreamThread-1" org.apache.kafka.streams.errors.StreamsException: stream-thread [StreamThread-1] Failed to rebalance at org.apache.kafka.streams.processor.internals.StreamThread.runLoop(StreamThread.java:612) at org.apache.kafka.streams.processor.internals.StreamThread.run(StreamThread.java:368) Caused by: java.lang.IndexOutOfBoundsException at java.nio.Buffer.checkIndex(Buffer.java:546) at java.nio.HeapByteBuffer.getLong(HeapByteBuffer.java:416) at org.apache.kafka.streams.kstream.internals.SessionKeySerde.extractEnd(SessionKeySerde.java:117) at org.apache.kafka.streams.state.internals.SessionKeySchema.segmentTimestamp(SessionKeySchema.java:45) at org.apache.kafka.streams.state.internals.RocksDBSegmentedBytesStore.put(RocksDBSegmentedBytesStore.java:71) at org.apache.kafka.streams.state.internals.RocksDBSegmentedBytesStore$1.restore(RocksDBSegmentedBytesStore.java:104) at org.apache.kafka.streams.processor.internals.ProcessorStateManager.restoreActiveState(ProcessorStateManager.java:230) at org.apache.kafka.streams.processor.internals.ProcessorStateManager.register(ProcessorStateManager.java:193) at org.apache.kafka.streams.processor.internals.AbstractProcessorContext.register(AbstractProcessorContext.java:99) at org.apache.kafka.streams.state.internals.RocksDBSegmentedBytesStore.init(RocksDBSegmentedBytesStore.java:101) at org.apache.kafka.streams.state.internals.ChangeLoggingSegmentedBytesStore.init(ChangeLoggingSegmentedBytesStore.java:68) at org.apache.kafka.streams.state.internals.MeteredSegmentedBytesStore.init(MeteredSegmentedBytesStore.java:66) at org.apache.kafka.streams.state.internals.RocksDBSessionStore.init(RocksDBSessionStore.java:78) at org.apache.kafka.streams.state.internals.CachingSessionStore.init(CachingSessionStore.java:97) at org.apache.kafka.streams.processor.internals.AbstractTask.initializeStateStores(AbstractTask.java:86) at org.apache.kafka.streams.processor.internals.StreamTask.<init>(StreamTask.java:141) at org.apache.kafka.streams.processor.internals.StreamThread.createStreamTask(StreamThread.java:834) at org.apache.kafka.streams.processor.internals.StreamThread$TaskCreator.createTask(StreamThread.java:1207) at org.apache.kafka.streams.processor.internals.StreamThread$AbstractTaskCreator.retryWithBackoff(StreamThread.java:1180) at org.apache.kafka.streams.processor.internals.StreamThread.addStreamTasks(StreamThread.java:937) at org.apache.kafka.streams.processor.internals.StreamThread.access$500(StreamThread.java:69) at org.apache.kafka.streams.processor.internals.StreamThread$1.onPartitionsAssigned(StreamThread.java:236) at org.apache.kafka.clients.consumer.internals.ConsumerCoordinator.onJoinComplete(ConsumerCoordinator.java:255) at org.apache.kafka.clients.consumer.internals.AbstractCoordinator.joinGroupIfNeeded(AbstractCoordinator.java:339) at org.apache.kafka.clients.consumer.internals.AbstractCoordinator.ensureActiveGroup(AbstractCoordinator.java:303) at org.apache.kafka.clients.consumer.internals.ConsumerCoordinator.poll(ConsumerCoordinator.java:286) at org.apache.kafka.clients.consumer.KafkaConsumer.pollOnce(KafkaConsumer.java:1030) at org.apache.kafka.clients.consumer.KafkaConsumer.poll(KafkaConsumer.java:995) at org.apache.kafka.streams.processor.internals.StreamThread.runLoop(StreamThread.java:582) ... 1 more

Can you please help me with this?

Thanks.
Marco

mjsax · 2017-02-21T22:37:45Z

@MarcoAbi Could you please report this at the Kafka mailing list: http://kafka.apache.org/contact

Thanks.

This is a follow up of apache/kafka#2166 - refactoring the store hierarchies as requested Author: Damian Guy <damian.guy@gmail.com> Reviewers: Guozhang Wang <wangguoz@gmail.com> Closes #2360 from dguy/state-store-refactor

This is a refactoring follow-up of apache/kafka#2166. Main refactoring changes: 1. Extract `InMemoryKeyValueStore` out of `InMemoryKeyValueStoreSupplier` and remove its duplicates in test package. 2. Add two abstract classes `AbstractKeyValueIterator` and `AbstractKeyValueStore` to collapse common functional logics. 3. Added specialized `BytesXXStore` to accommodate cases where key value types are Bytes / byte[] so that we can save calling the dummy serdes. 4. Make the key type in `ThreadCache` from byte[] to Bytes, as SessionStore / WindowStore's result serialized bytes are in the form of Bytes anyways, so that we can save unnecessary `Bytes.get()` and `Bytes.wrap(bytes)`. Each of these should arguably be a separate PR and I apologize for the mess, this is because this branch was extracted from a rather large diff that has multiple refactoring mingled together and dguy and myself have already put lots of efforts to break it down to a few separate PRs, and this is the only left-over work. Such PR won't happen in the future. Ping dguy enothereska mjsax for reviews Author: Guozhang Wang <wangguoz@gmail.com> Reviewers: Damian Guy, Matthias J. Sax, Jun Rao Closes #2333 from guozhangwang/K3452-followup-state-store-refactor

session windows

1876df6

enothereska reviewed Nov 28, 2016

View reviewed changes

added comments as per feedback from eno

b28bdef

mjsax reviewed Nov 30, 2016

View reviewed changes

dguy added 7 commits November 30, 2016 10:42

address comments

05198c6

rename getter on SessionWindows

b6ce6ca

change the session gap semantics

5d4d003

add sessionWindows method to PersistentKeyValueFactory

7538adc

rename SessionMerger -> Merger

38abef1

merge with trunk

e7b6193

merge trunk

6e00e70

fix checkstyle

d066168

enothereska reviewed Dec 14, 2016

View reviewed changes

dguy added 2 commits December 14, 2016 14:06

comments

4171d41

merge trunk and comments

090bace

enothereska reviewed Dec 14, 2016

View reviewed changes

clear cache on close. address some feedback

f1831ad

enothereska reviewed Dec 14, 2016

View reviewed changes

dguy added 2 commits January 5, 2017 11:32

address most comments

72b9dae

address comments

a6cf769

minor refactoring to remove some duplication

c9ea501

asfgit closed this in e0de3a4 Jan 6, 2017

guozhangwang mentioned this pull request Jan 9, 2017

KAFKA-3452 Follow-up: Optimize ByteStore Scenarios #2333

Closed

This was referenced Jan 10, 2017

MINOR: rename SessionStore.findSessionsToMerge to findSessions #2339

Closed

KAFKA-3452: follow up - fix state store restoration for session and window stores #2359

Closed

KAFKA-3452 Follow-up: Refactoring StateStore hierarchies #2360

Closed

dguy deleted the kafka-3452-session-merge branch January 13, 2017 08:39

mjsax mentioned this pull request Mar 4, 2026

KAFKA-20134: Implement TimestampedWindowStoreWithHeaders (N/N) #21581

Merged

		import org.apache.kafka.streams.kstream.Windowed;

		import java.nio.ByteBuffer;



		/**
		* A session based window specification uses for aggregating events into sessions.

Conversation

dguy commented Nov 24, 2016

Uh oh!

dguy commented Nov 24, 2016

Uh oh!

enothereska 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!

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!

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!

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!

asfbot commented Dec 12, 2016

Uh oh!

asfbot commented Dec 12, 2016

Uh oh!

asfbot commented Dec 12, 2016

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!