kafka-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From guozh...@apache.org
Subject kafka git commit: KAFKA-3452; Follow-up: Add SessionWindows
Date Thu, 12 Jan 2017 04:33:45 GMT
Repository: kafka
Updated Branches:
  refs/heads/trunk 23dff4b04 -> 3c6051165


KAFKA-3452; Follow-up: Add SessionWindows

 - TimeWindows represent half-open time intervals while SessionWindows represent closed time intervals

Author: Matthias J. Sax <matthias@confluent.io>

Reviewers: Damian Guy <damian.guy@gmail.com>, Guozhang Wang <wangguoz@gmail.com>

Closes #2342 from mjsax/kafka-3452-session-window-follow-up


Project: http://git-wip-us.apache.org/repos/asf/kafka/repo
Commit: http://git-wip-us.apache.org/repos/asf/kafka/commit/3c605116
Tree: http://git-wip-us.apache.org/repos/asf/kafka/tree/3c605116
Diff: http://git-wip-us.apache.org/repos/asf/kafka/diff/3c605116

Branch: refs/heads/trunk
Commit: 3c60511655df3349323e394bdc3836300991352f
Parents: 23dff4b
Author: Matthias J. Sax <matthias@confluent.io>
Authored: Wed Jan 11 20:33:42 2017 -0800
Committer: Guozhang Wang <wangguoz@gmail.com>
Committed: Wed Jan 11 20:33:42 2017 -0800

----------------------------------------------------------------------
 .../apache/kafka/streams/kstream/Window.java    |  21 +++-
 .../KStreamSessionWindowAggregate.java          |  12 +-
 .../kstream/internals/SessionKeySerde.java      |   2 +-
 .../kstream/internals/SessionWindow.java        |  58 +++++++++
 .../streams/kstream/internals/TimeWindow.java   |   7 +-
 .../kstream/internals/UnlimitedWindow.java      |   6 +-
 .../state/internals/SessionKeySchema.java       |   6 +-
 .../KStreamAggregationIntegrationTest.java      |  34 ++---
 .../internals/KGroupedStreamImplTest.java       |  18 +--
 ...reamSessionWindowAggregateProcessorTest.java |  40 +++---
 .../kstream/internals/SessionKeySerdeTest.java  |  10 +-
 .../kstream/internals/SessionWindowTest.java    | 124 +++++++++++++++++++
 .../internals/CachingSessionStoreTest.java      |  42 +++----
 .../RocksDBSegmentedBytesStoreTest.java         |  44 +++----
 .../internals/RocksDBSessionStoreTest.java      |  46 +++----
 15 files changed, 335 insertions(+), 135 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/kafka/blob/3c605116/streams/src/main/java/org/apache/kafka/streams/kstream/Window.java
----------------------------------------------------------------------
diff --git a/streams/src/main/java/org/apache/kafka/streams/kstream/Window.java b/streams/src/main/java/org/apache/kafka/streams/kstream/Window.java
index 7d78d74..13a9529 100644
--- a/streams/src/main/java/org/apache/kafka/streams/kstream/Window.java
+++ b/streams/src/main/java/org/apache/kafka/streams/kstream/Window.java
@@ -22,16 +22,27 @@ package org.apache.kafka.streams.kstream;
  */
 public abstract class Window {
 
-    private long start;
-    private long end;
+    protected final long start;
+    protected final long end;
 
     /**
      * Create a new window for the given start time (inclusive) and end time (exclusive).
      *
      * @param start  the start timestamp of the window (inclusive)
      * @param end    the end timestamp of the window (exclusive)
+     * @throws IllegalArgumentException if {@code start} or {@code end} is negative or if {@code end} is smaller than
+     * {@code start}
      */
-    public Window(long start, long end) {
+    public Window(long start, long end) throws IllegalArgumentException {
+        if (start < 0) {
+            throw new IllegalArgumentException("Window start time cannot be negative.");
+        }
+        if (end < 0) {
+            throw new IllegalArgumentException("Window end time cannot be negative.");
+        }
+        if (end < start) {
+            throw new IllegalArgumentException("Window end time cannot be smaller than window start time.");
+        }
         this.start = start;
         this.end = end;
     }
@@ -56,9 +67,7 @@ public abstract class Window {
      * @param other  another window
      * @return       {@code true} if {@code other} overlaps with this window&mdash;{@code false} otherwise
      */
-    public boolean overlap(Window other) {
-        return this.start() < other.end() || other.start() < this.end();
-    }
+    public abstract boolean overlap(Window other);
 
     @Override
     public boolean equals(Object obj) {

http://git-wip-us.apache.org/repos/asf/kafka/blob/3c605116/streams/src/main/java/org/apache/kafka/streams/kstream/internals/KStreamSessionWindowAggregate.java
----------------------------------------------------------------------
diff --git a/streams/src/main/java/org/apache/kafka/streams/kstream/internals/KStreamSessionWindowAggregate.java b/streams/src/main/java/org/apache/kafka/streams/kstream/internals/KStreamSessionWindowAggregate.java
index bb86f52..70b2b90 100644
--- a/streams/src/main/java/org/apache/kafka/streams/kstream/internals/KStreamSessionWindowAggregate.java
+++ b/streams/src/main/java/org/apache/kafka/streams/kstream/internals/KStreamSessionWindowAggregate.java
@@ -88,8 +88,8 @@ class KStreamSessionWindowAggregate<K, V, T> implements KStreamAggProcessorSuppl
 
             final long timestamp = context().timestamp();
             final List<KeyValue<Windowed<K>, T>> merged = new ArrayList<>();
-            final TimeWindow newTimeWindow = new TimeWindow(timestamp, timestamp);
-            TimeWindow mergedWindow = newTimeWindow;
+            final SessionWindow newSessionWindow = new SessionWindow(timestamp, timestamp);
+            SessionWindow mergedWindow = newSessionWindow;
             T agg = initializer.apply();
 
             try (final KeyValueIterator<Windowed<K>, T> iterator = store.findSessions(key, timestamp - windows.inactivityGap(),
@@ -98,13 +98,13 @@ class KStreamSessionWindowAggregate<K, V, T> implements KStreamAggProcessorSuppl
                     final KeyValue<Windowed<K>, T> next = iterator.next();
                     merged.add(next);
                     agg = sessionMerger.apply(key, agg, next.value);
-                    mergedWindow = mergeTimeWindow(mergedWindow, (TimeWindow) next.key.window());
+                    mergedWindow = mergeSessionWindow(mergedWindow, (SessionWindow) next.key.window());
                 }
             }
 
             agg = aggregator.apply(key, value, agg);
             final Windowed<K> sessionKey = new Windowed<>(key, mergedWindow);
-            if (!mergedWindow.equals(newTimeWindow)) {
+            if (!mergedWindow.equals(newSessionWindow)) {
                 for (final KeyValue<Windowed<K>, T> session : merged) {
                     store.remove(session.key);
                     tupleForwarder.maybeForward(session.key, null, session.value);
@@ -117,10 +117,10 @@ class KStreamSessionWindowAggregate<K, V, T> implements KStreamAggProcessorSuppl
     }
 
 
-    private TimeWindow mergeTimeWindow(final TimeWindow one, final TimeWindow two) {
+    private SessionWindow mergeSessionWindow(final SessionWindow one, final SessionWindow two) {
         final long start = one.start() < two.start() ? one.start() : two.start();
         final long end = one.end() > two.end() ? one.end() : two.end();
-        return new TimeWindow(start, end);
+        return new SessionWindow(start, end);
     }
 
     @Override

http://git-wip-us.apache.org/repos/asf/kafka/blob/3c605116/streams/src/main/java/org/apache/kafka/streams/kstream/internals/SessionKeySerde.java
----------------------------------------------------------------------
diff --git a/streams/src/main/java/org/apache/kafka/streams/kstream/internals/SessionKeySerde.java b/streams/src/main/java/org/apache/kafka/streams/kstream/internals/SessionKeySerde.java
index 165d5c6..48213d6 100644
--- a/streams/src/main/java/org/apache/kafka/streams/kstream/internals/SessionKeySerde.java
+++ b/streams/src/main/java/org/apache/kafka/streams/kstream/internals/SessionKeySerde.java
@@ -131,7 +131,7 @@ public class SessionKeySerde<K> implements Serde<Windowed<K>> {
         final ByteBuffer buffer = ByteBuffer.wrap(binaryKey);
         final long start = buffer.getLong(binaryKey.length - TIMESTAMP_SIZE);
         final long end = buffer.getLong(binaryKey.length - 2 * TIMESTAMP_SIZE);
-        return new Windowed<>(key, new TimeWindow(start, end));
+        return new Windowed<>(key, new SessionWindow(start, end));
     }
 
     private static <K> K extractKey(final byte[] binaryKey, Deserializer<K> deserializer) {

http://git-wip-us.apache.org/repos/asf/kafka/blob/3c605116/streams/src/main/java/org/apache/kafka/streams/kstream/internals/SessionWindow.java
----------------------------------------------------------------------
diff --git a/streams/src/main/java/org/apache/kafka/streams/kstream/internals/SessionWindow.java b/streams/src/main/java/org/apache/kafka/streams/kstream/internals/SessionWindow.java
new file mode 100644
index 0000000..db63029
--- /dev/null
+++ b/streams/src/main/java/org/apache/kafka/streams/kstream/internals/SessionWindow.java
@@ -0,0 +1,58 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ * <p>
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * <p>
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.kafka.streams.kstream.internals;
+
+import org.apache.kafka.streams.kstream.Window;
+
+/**
+ * A session window covers a closed time interval with its start and end timestamp both being an inclusive boundary.
+ *
+ * @see TimeWindow
+ * @see UnlimitedWindow
+ * @see org.apache.kafka.streams.kstream.SessionWindows
+ */
+public final class SessionWindow extends Window {
+
+    /**
+     * Create a new window for the given start time and end time (both inclusive).
+     *
+     * @param start  the start timestamp of the window
+     * @param end    the end timestamp of the window
+     */
+    public SessionWindow(final long start, final long end) {
+        super(start, end);
+    }
+
+    /**
+     * Check if the given window overlaps with this window.
+     *
+     * @param other  another window
+     * @return       {@code true} if {@code other} overlaps with this window&mdash;{@code false} otherwise
+     * @throws IllegalArgumentException if the {@code other} window has a different type than {@link this} window
+     */
+    public boolean overlap(final Window other) throws IllegalArgumentException {
+        if (getClass() != other.getClass()) {
+            throw new IllegalArgumentException("Cannot compare windows of different type. Other window has type "
+                + other.getClass());
+        }
+        final SessionWindow otherWindow = (SessionWindow) other;
+        return !(otherWindow.end < start || end < otherWindow.start);
+    }
+
+}

http://git-wip-us.apache.org/repos/asf/kafka/blob/3c605116/streams/src/main/java/org/apache/kafka/streams/kstream/internals/TimeWindow.java
----------------------------------------------------------------------
diff --git a/streams/src/main/java/org/apache/kafka/streams/kstream/internals/TimeWindow.java b/streams/src/main/java/org/apache/kafka/streams/kstream/internals/TimeWindow.java
index 5dfb9eb..630821f 100644
--- a/streams/src/main/java/org/apache/kafka/streams/kstream/internals/TimeWindow.java
+++ b/streams/src/main/java/org/apache/kafka/streams/kstream/internals/TimeWindow.java
@@ -27,7 +27,12 @@ public class TimeWindow extends Window {
 
     @Override
     public boolean overlap(Window other) {
-        return getClass() == other.getClass() && super.overlap(other);
+        if (getClass() != other.getClass()) {
+            throw new IllegalArgumentException("Cannot compare windows of different type. Other window has type "
+                + other.getClass());
+        }
+        final TimeWindow otherWindow = (TimeWindow) other;
+        return start < otherWindow.end && otherWindow.start < end;
     }
 
 }
\ No newline at end of file

http://git-wip-us.apache.org/repos/asf/kafka/blob/3c605116/streams/src/main/java/org/apache/kafka/streams/kstream/internals/UnlimitedWindow.java
----------------------------------------------------------------------
diff --git a/streams/src/main/java/org/apache/kafka/streams/kstream/internals/UnlimitedWindow.java b/streams/src/main/java/org/apache/kafka/streams/kstream/internals/UnlimitedWindow.java
index 4b93f9b..e9ec040 100644
--- a/streams/src/main/java/org/apache/kafka/streams/kstream/internals/UnlimitedWindow.java
+++ b/streams/src/main/java/org/apache/kafka/streams/kstream/internals/UnlimitedWindow.java
@@ -27,7 +27,11 @@ public class UnlimitedWindow extends Window {
 
     @Override
     public boolean overlap(Window other) {
-        return getClass() == other.getClass() && super.overlap(other);
+        if (getClass() != other.getClass()) {
+            throw new IllegalArgumentException("Cannot compare windows of different type. Other window has type "
+                + other.getClass());
+        }
+        return true;
     }
 
 }
\ No newline at end of file

http://git-wip-us.apache.org/repos/asf/kafka/blob/3c605116/streams/src/main/java/org/apache/kafka/streams/state/internals/SessionKeySchema.java
----------------------------------------------------------------------
diff --git a/streams/src/main/java/org/apache/kafka/streams/state/internals/SessionKeySchema.java b/streams/src/main/java/org/apache/kafka/streams/state/internals/SessionKeySchema.java
index b15eec9..604abb3 100644
--- a/streams/src/main/java/org/apache/kafka/streams/state/internals/SessionKeySchema.java
+++ b/streams/src/main/java/org/apache/kafka/streams/state/internals/SessionKeySchema.java
@@ -20,7 +20,7 @@ import org.apache.kafka.common.serialization.Serdes;
 import org.apache.kafka.common.utils.Bytes;
 import org.apache.kafka.streams.kstream.Windowed;
 import org.apache.kafka.streams.kstream.internals.SessionKeySerde;
-import org.apache.kafka.streams.kstream.internals.TimeWindow;
+import org.apache.kafka.streams.kstream.internals.SessionWindow;
 import org.apache.kafka.streams.state.KeyValueIterator;
 
 import java.util.List;
@@ -30,13 +30,13 @@ class SessionKeySchema implements SegmentedBytesStore.KeySchema {
 
     @Override
     public Bytes upperRange(final Bytes key, final long to) {
-        final Windowed<Bytes> sessionKey = new Windowed<>(key, new TimeWindow(to, Long.MAX_VALUE));
+        final Windowed<Bytes> sessionKey = new Windowed<>(key, new SessionWindow(to, Long.MAX_VALUE));
         return SessionKeySerde.toBinary(sessionKey, Serdes.Bytes().serializer());
     }
 
     @Override
     public Bytes lowerRange(final Bytes key, final long from) {
-        final Windowed<Bytes> sessionKey = new Windowed<>(key, new TimeWindow(0, Math.max(0, from)));
+        final Windowed<Bytes> sessionKey = new Windowed<>(key, new SessionWindow(0, Math.max(0, from)));
         return SessionKeySerde.toBinary(sessionKey, Serdes.Bytes().serializer());
     }
 

http://git-wip-us.apache.org/repos/asf/kafka/blob/3c605116/streams/src/test/java/org/apache/kafka/streams/integration/KStreamAggregationIntegrationTest.java
----------------------------------------------------------------------
diff --git a/streams/src/test/java/org/apache/kafka/streams/integration/KStreamAggregationIntegrationTest.java b/streams/src/test/java/org/apache/kafka/streams/integration/KStreamAggregationIntegrationTest.java
index 5cc2a59..0833f3c 100644
--- a/streams/src/test/java/org/apache/kafka/streams/integration/KStreamAggregationIntegrationTest.java
+++ b/streams/src/test/java/org/apache/kafka/streams/integration/KStreamAggregationIntegrationTest.java
@@ -39,7 +39,7 @@ import org.apache.kafka.streams.kstream.Reducer;
 import org.apache.kafka.streams.kstream.SessionWindows;
 import org.apache.kafka.streams.kstream.TimeWindows;
 import org.apache.kafka.streams.kstream.Windowed;
-import org.apache.kafka.streams.kstream.internals.TimeWindow;
+import org.apache.kafka.streams.kstream.internals.SessionWindow;
 import org.apache.kafka.streams.state.KeyValueIterator;
 import org.apache.kafka.streams.state.QueryableStoreTypes;
 import org.apache.kafka.streams.state.ReadOnlySessionStore;
@@ -506,13 +506,13 @@ public class KStreamAggregationIntegrationTest {
 
         startStreams();
         latch.await(30, TimeUnit.SECONDS);
-        assertThat(results.get(new Windowed<>("bob", new TimeWindow(t1, t1))), equalTo(1L));
-        assertThat(results.get(new Windowed<>("penny", new TimeWindow(t1, t1))), equalTo(1L));
-        assertThat(results.get(new Windowed<>("jo", new TimeWindow(t1, t1))), equalTo(1L));
-        assertThat(results.get(new Windowed<>("jo", new TimeWindow(t4, t4))), equalTo(1L));
-        assertThat(results.get(new Windowed<>("emily", new TimeWindow(t1, t2))), equalTo(2L));
-        assertThat(results.get(new Windowed<>("bob", new TimeWindow(t3, t4))), equalTo(2L));
-        assertThat(results.get(new Windowed<>("penny", new TimeWindow(t3, t3))), equalTo(1L));
+        assertThat(results.get(new Windowed<>("bob", new SessionWindow(t1, t1))), equalTo(1L));
+        assertThat(results.get(new Windowed<>("penny", new SessionWindow(t1, t1))), equalTo(1L));
+        assertThat(results.get(new Windowed<>("jo", new SessionWindow(t1, t1))), equalTo(1L));
+        assertThat(results.get(new Windowed<>("jo", new SessionWindow(t4, t4))), equalTo(1L));
+        assertThat(results.get(new Windowed<>("emily", new SessionWindow(t1, t2))), equalTo(2L));
+        assertThat(results.get(new Windowed<>("bob", new SessionWindow(t3, t4))), equalTo(2L));
+        assertThat(results.get(new Windowed<>("penny", new SessionWindow(t3, t3))), equalTo(1L));
     }
 
     @Test
@@ -601,18 +601,18 @@ public class KStreamAggregationIntegrationTest {
                 = kafkaStreams.store(userSessionsStore, QueryableStoreTypes.<String, String>sessionStore());
 
         // verify correct data received
-        assertThat(results.get(new Windowed<>("bob", new TimeWindow(t1, t1))), equalTo("start"));
-        assertThat(results.get(new Windowed<>("penny", new TimeWindow(t1, t1))), equalTo("start"));
-        assertThat(results.get(new Windowed<>("jo", new TimeWindow(t1, t1))), equalTo("pause"));
-        assertThat(results.get(new Windowed<>("jo", new TimeWindow(t4, t4))), equalTo("resume"));
-        assertThat(results.get(new Windowed<>("emily", new TimeWindow(t1, t2))), equalTo("pause:resume"));
-        assertThat(results.get(new Windowed<>("bob", new TimeWindow(t3, t4))), equalTo("pause:resume"));
-        assertThat(results.get(new Windowed<>("penny", new TimeWindow(t3, t3))), equalTo("stop"));
+        assertThat(results.get(new Windowed<>("bob", new SessionWindow(t1, t1))), equalTo("start"));
+        assertThat(results.get(new Windowed<>("penny", new SessionWindow(t1, t1))), equalTo("start"));
+        assertThat(results.get(new Windowed<>("jo", new SessionWindow(t1, t1))), equalTo("pause"));
+        assertThat(results.get(new Windowed<>("jo", new SessionWindow(t4, t4))), equalTo("resume"));
+        assertThat(results.get(new Windowed<>("emily", new SessionWindow(t1, t2))), equalTo("pause:resume"));
+        assertThat(results.get(new Windowed<>("bob", new SessionWindow(t3, t4))), equalTo("pause:resume"));
+        assertThat(results.get(new Windowed<>("penny", new SessionWindow(t3, t3))), equalTo("stop"));
 
         // verify can query data via IQ
         final KeyValueIterator<Windowed<String>, String> bob = sessionStore.fetch("bob");
-        assertThat(bob.next(), equalTo(KeyValue.pair(new Windowed<>("bob", new TimeWindow(t1, t1)), "start")));
-        assertThat(bob.next(), equalTo(KeyValue.pair(new Windowed<>("bob", new TimeWindow(t3, t4)), "pause:resume")));
+        assertThat(bob.next(), equalTo(KeyValue.pair(new Windowed<>("bob", new SessionWindow(t1, t1)), "start")));
+        assertThat(bob.next(), equalTo(KeyValue.pair(new Windowed<>("bob", new SessionWindow(t3, t4)), "pause:resume")));
         assertFalse(bob.hasNext());
 
     }

http://git-wip-us.apache.org/repos/asf/kafka/blob/3c605116/streams/src/test/java/org/apache/kafka/streams/kstream/internals/KGroupedStreamImplTest.java
----------------------------------------------------------------------
diff --git a/streams/src/test/java/org/apache/kafka/streams/kstream/internals/KGroupedStreamImplTest.java b/streams/src/test/java/org/apache/kafka/streams/kstream/internals/KGroupedStreamImplTest.java
index 62dd1d5..729e190 100644
--- a/streams/src/test/java/org/apache/kafka/streams/kstream/internals/KGroupedStreamImplTest.java
+++ b/streams/src/test/java/org/apache/kafka/streams/kstream/internals/KGroupedStreamImplTest.java
@@ -173,9 +173,9 @@ public class KGroupedStreamImplTest {
         driver.setTime(100);
         driver.process(TOPIC, "1", "1");
         driver.flushState();
-        assertEquals(Integer.valueOf(2), results.get(new Windowed<>("1", new TimeWindow(10, 30))));
-        assertEquals(Integer.valueOf(1), results.get(new Windowed<>("2", new TimeWindow(15, 15))));
-        assertEquals(Integer.valueOf(3), results.get(new Windowed<>("1", new TimeWindow(70, 100))));
+        assertEquals(Integer.valueOf(2), results.get(new Windowed<>("1", new SessionWindow(10, 30))));
+        assertEquals(Integer.valueOf(1), results.get(new Windowed<>("2", new SessionWindow(15, 15))));
+        assertEquals(Integer.valueOf(3), results.get(new Windowed<>("1", new SessionWindow(70, 100))));
     }
 
     @Test
@@ -202,9 +202,9 @@ public class KGroupedStreamImplTest {
         driver.setTime(100);
         driver.process(TOPIC, "1", "1");
         driver.flushState();
-        assertEquals(Long.valueOf(2), results.get(new Windowed<>("1", new TimeWindow(10, 30))));
-        assertEquals(Long.valueOf(1), results.get(new Windowed<>("2", new TimeWindow(15, 15))));
-        assertEquals(Long.valueOf(3), results.get(new Windowed<>("1", new TimeWindow(70, 100))));
+        assertEquals(Long.valueOf(2), results.get(new Windowed<>("1", new SessionWindow(10, 30))));
+        assertEquals(Long.valueOf(1), results.get(new Windowed<>("2", new SessionWindow(15, 15))));
+        assertEquals(Long.valueOf(3), results.get(new Windowed<>("1", new SessionWindow(70, 100))));
     }
 
     @Test
@@ -238,9 +238,9 @@ public class KGroupedStreamImplTest {
         driver.setTime(100);
         driver.process(TOPIC, "1", "C");
         driver.flushState();
-        assertEquals("A:B", results.get(new Windowed<>("1", new TimeWindow(10, 30))));
-        assertEquals("Z", results.get(new Windowed<>("2", new TimeWindow(15, 15))));
-        assertEquals("A:B:C", results.get(new Windowed<>("1", new TimeWindow(70, 100))));
+        assertEquals("A:B", results.get(new Windowed<>("1", new SessionWindow(10, 30))));
+        assertEquals("Z", results.get(new Windowed<>("2", new SessionWindow(15, 15))));
+        assertEquals("A:B:C", results.get(new Windowed<>("1", new SessionWindow(70, 100))));
     }
 
     @Test(expected = NullPointerException.class)

http://git-wip-us.apache.org/repos/asf/kafka/blob/3c605116/streams/src/test/java/org/apache/kafka/streams/kstream/internals/KStreamSessionWindowAggregateProcessorTest.java
----------------------------------------------------------------------
diff --git a/streams/src/test/java/org/apache/kafka/streams/kstream/internals/KStreamSessionWindowAggregateProcessorTest.java b/streams/src/test/java/org/apache/kafka/streams/kstream/internals/KStreamSessionWindowAggregateProcessorTest.java
index 0ecaf3a..c3368a1 100644
--- a/streams/src/test/java/org/apache/kafka/streams/kstream/internals/KStreamSessionWindowAggregateProcessorTest.java
+++ b/streams/src/test/java/org/apache/kafka/streams/kstream/internals/KStreamSessionWindowAggregateProcessorTest.java
@@ -184,9 +184,9 @@ public class KStreamSessionWindowAggregateProcessorTest {
 
         sessionStore.flush();
         assertEquals(Arrays.asList(
-                KeyValue.pair(new Windowed<>(sessionId, new TimeWindow(0, 0)), new Change<>(1L, null)),
-                KeyValue.pair(new Windowed<>(sessionId, new TimeWindow(GAP_MS + 1, GAP_MS + 1)), new Change<>(2L, null)),
-                KeyValue.pair(new Windowed<>(sessionId, new TimeWindow(time, time)), new Change<>(3L, null))
+                KeyValue.pair(new Windowed<>(sessionId, new SessionWindow(0, 0)), new Change<>(1L, null)),
+                KeyValue.pair(new Windowed<>(sessionId, new SessionWindow(GAP_MS + 1, GAP_MS + 1)), new Change<>(2L, null)),
+                KeyValue.pair(new Windowed<>(sessionId, new SessionWindow(time, time)), new Change<>(3L, null))
 
         ), results);
 
@@ -200,14 +200,14 @@ public class KStreamSessionWindowAggregateProcessorTest {
 
         // first ensure it is in the store
         final KeyValueIterator<Windowed<String>, Long> a1 = sessionStore.findSessions("a", 0, 0);
-        assertEquals(KeyValue.pair(new Windowed<>("a", new TimeWindow(0, 0)), 1L), a1.next());
+        assertEquals(KeyValue.pair(new Windowed<>("a", new SessionWindow(0, 0)), 1L), a1.next());
 
         context.setTime(100);
         processor.process("a", "2");
         // a1 from above should have been removed
         // should have merged session in store
         final KeyValueIterator<Windowed<String>, Long> a2 = sessionStore.findSessions("a", 0, 100);
-        assertEquals(KeyValue.pair(new Windowed<>("a", new TimeWindow(0, 100)), 2L), a2.next());
+        assertEquals(KeyValue.pair(new Windowed<>("a", new SessionWindow(0, 100)), 2L), a2.next());
         assertFalse(a2.hasNext());
     }
 
@@ -229,13 +229,13 @@ public class KStreamSessionWindowAggregateProcessorTest {
 
         sessionStore.flush();
 
-        assertEquals(Arrays.asList(KeyValue.pair(new Windowed<>("a", new TimeWindow(0, 0)), new Change<>(1L, null)),
-                                   KeyValue.pair(new Windowed<>("b", new TimeWindow(0, 0)), new Change<>(1L, null)),
-                                   KeyValue.pair(new Windowed<>("c", new TimeWindow(0, 0)), new Change<>(1L, null)),
-                                   KeyValue.pair(new Windowed<>("d", new TimeWindow(0, GAP_MS / 2)), new Change<>(2L, null)),
-                                   KeyValue.pair(new Windowed<>("b", new TimeWindow(GAP_MS + 1, GAP_MS + 1)), new Change<>(1L, null)),
-                                   KeyValue.pair(new Windowed<>("a", new TimeWindow(GAP_MS + 1, GAP_MS + 1 + GAP_MS / 2)), new Change<>(2L, null)),
-                                   KeyValue.pair(new Windowed<>("c", new TimeWindow(GAP_MS + 1 + GAP_MS / 2, GAP_MS + 1 + GAP_MS / 2)), new Change<>(1L, null))
+        assertEquals(Arrays.asList(KeyValue.pair(new Windowed<>("a", new SessionWindow(0, 0)), new Change<>(1L, null)),
+                                   KeyValue.pair(new Windowed<>("b", new SessionWindow(0, 0)), new Change<>(1L, null)),
+                                   KeyValue.pair(new Windowed<>("c", new SessionWindow(0, 0)), new Change<>(1L, null)),
+                                   KeyValue.pair(new Windowed<>("d", new SessionWindow(0, GAP_MS / 2)), new Change<>(2L, null)),
+                                   KeyValue.pair(new Windowed<>("b", new SessionWindow(GAP_MS + 1, GAP_MS + 1)), new Change<>(1L, null)),
+                                   KeyValue.pair(new Windowed<>("a", new SessionWindow(GAP_MS + 1, GAP_MS + 1 + GAP_MS / 2)), new Change<>(2L, null)),
+                                   KeyValue.pair(new Windowed<>("c", new SessionWindow(GAP_MS + 1 + GAP_MS / 2, GAP_MS + 1 + GAP_MS / 2)), new Change<>(1L, null))
                      ),
                      results);
     }
@@ -250,8 +250,8 @@ public class KStreamSessionWindowAggregateProcessorTest {
         context.setTime(GAP_MS + 1);
         processor.process("a", "1");
         processor.process("a", "2");
-        final long t0 = getter.get(new Windowed<>("a", new TimeWindow(0, 0)));
-        final long t1 = getter.get(new Windowed<>("a", new TimeWindow(GAP_MS + 1, GAP_MS + 1)));
+        final long t0 = getter.get(new Windowed<>("a", new SessionWindow(0, 0)));
+        final long t1 = getter.get(new Windowed<>("a", new SessionWindow(GAP_MS + 1, GAP_MS + 1)));
         assertEquals(1L, t0);
         assertEquals(2L, t1);
     }
@@ -266,9 +266,9 @@ public class KStreamSessionWindowAggregateProcessorTest {
         processor.process("b", "1");
         processor.process("c", "1");
 
-        assertEquals(Arrays.asList(KeyValue.pair(new Windowed<>("a", new TimeWindow(0, 0)), new Change<>(1L, null)),
-                                   KeyValue.pair(new Windowed<>("b", new TimeWindow(0, 0)), new Change<>(1L, null)),
-                                   KeyValue.pair(new Windowed<>("c", new TimeWindow(0, 0)), new Change<>(1L, null))), results);
+        assertEquals(Arrays.asList(KeyValue.pair(new Windowed<>("a", new SessionWindow(0, 0)), new Change<>(1L, null)),
+                                   KeyValue.pair(new Windowed<>("b", new SessionWindow(0, 0)), new Change<>(1L, null)),
+                                   KeyValue.pair(new Windowed<>("c", new SessionWindow(0, 0)), new Change<>(1L, null))), results);
     }
 
     @Test
@@ -280,9 +280,9 @@ public class KStreamSessionWindowAggregateProcessorTest {
         processor.process("a", "1");
         context.setTime(5);
         processor.process("a", "1");
-        assertEquals(Arrays.asList(KeyValue.pair(new Windowed<>("a", new TimeWindow(0, 0)), new Change<>(1L, null)),
-                                   KeyValue.pair(new Windowed<>("a", new TimeWindow(0, 0)), new Change<>(null, null)),
-                                   KeyValue.pair(new Windowed<>("a", new TimeWindow(0, 5)), new Change<>(2L, null))), results);
+        assertEquals(Arrays.asList(KeyValue.pair(new Windowed<>("a", new SessionWindow(0, 0)), new Change<>(1L, null)),
+                                   KeyValue.pair(new Windowed<>("a", new SessionWindow(0, 0)), new Change<>(null, null)),
+                                   KeyValue.pair(new Windowed<>("a", new SessionWindow(0, 5)), new Change<>(2L, null))), results);
 
     }
 

http://git-wip-us.apache.org/repos/asf/kafka/blob/3c605116/streams/src/test/java/org/apache/kafka/streams/kstream/internals/SessionKeySerdeTest.java
----------------------------------------------------------------------
diff --git a/streams/src/test/java/org/apache/kafka/streams/kstream/internals/SessionKeySerdeTest.java b/streams/src/test/java/org/apache/kafka/streams/kstream/internals/SessionKeySerdeTest.java
index 2f5972c..3a0f490 100644
--- a/streams/src/test/java/org/apache/kafka/streams/kstream/internals/SessionKeySerdeTest.java
+++ b/streams/src/test/java/org/apache/kafka/streams/kstream/internals/SessionKeySerdeTest.java
@@ -30,7 +30,7 @@ public class SessionKeySerdeTest {
 
     @Test
     public void shouldSerializeDeserialize() throws Exception {
-        final Windowed<Long> key = new Windowed<>(1L, new TimeWindow(10, 100));
+        final Windowed<Long> key = new Windowed<>(1L, new SessionWindow(10, 100));
         final SessionKeySerde<Long> serde = new SessionKeySerde<>(Serdes.Long());
         final byte[] bytes = serde.serializer().serialize("t", key);
         final Windowed<Long> result = serde.deserializer().deserialize("t", bytes);
@@ -57,7 +57,7 @@ public class SessionKeySerdeTest {
 
     @Test
     public void shouldConvertToBinaryAndBack() throws Exception {
-        final Windowed<String> key = new Windowed<>("key", new TimeWindow(10, 20));
+        final Windowed<String> key = new Windowed<>("key", new SessionWindow(10, 20));
         final Bytes serialized = SessionKeySerde.toBinary(key, Serdes.String().serializer());
         final Windowed<String> result = SessionKeySerde.from(serialized.get(), Serdes.String().deserializer());
         assertEquals(key, result);
@@ -65,21 +65,21 @@ public class SessionKeySerdeTest {
 
     @Test
     public void shouldExtractEndTimeFromBinary() throws Exception {
-        final Windowed<String> key = new Windowed<>("key", new TimeWindow(10, 100));
+        final Windowed<String> key = new Windowed<>("key", new SessionWindow(10, 100));
         final Bytes serialized = SessionKeySerde.toBinary(key, Serdes.String().serializer());
         assertEquals(100, SessionKeySerde.extractEnd(serialized.get()));
     }
 
     @Test
     public void shouldExtractStartTimeFromBinary() throws Exception {
-        final Windowed<String> key = new Windowed<>("key", new TimeWindow(50, 100));
+        final Windowed<String> key = new Windowed<>("key", new SessionWindow(50, 100));
         final Bytes serialized = SessionKeySerde.toBinary(key, Serdes.String().serializer());
         assertEquals(50, SessionKeySerde.extractStart(serialized.get()));
     }
 
     @Test
     public void shouldExtractKeyBytesFromBinary() throws Exception {
-        final Windowed<String> key = new Windowed<>("blah", new TimeWindow(50, 100));
+        final Windowed<String> key = new Windowed<>("blah", new SessionWindow(50, 100));
         final Bytes serialized = SessionKeySerde.toBinary(key, Serdes.String().serializer());
         assertArrayEquals("blah".getBytes(), SessionKeySerde.extractKeyBytes(serialized.get()));
     }

http://git-wip-us.apache.org/repos/asf/kafka/blob/3c605116/streams/src/test/java/org/apache/kafka/streams/kstream/internals/SessionWindowTest.java
----------------------------------------------------------------------
diff --git a/streams/src/test/java/org/apache/kafka/streams/kstream/internals/SessionWindowTest.java b/streams/src/test/java/org/apache/kafka/streams/kstream/internals/SessionWindowTest.java
new file mode 100644
index 0000000..2df7741
--- /dev/null
+++ b/streams/src/test/java/org/apache/kafka/streams/kstream/internals/SessionWindowTest.java
@@ -0,0 +1,124 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ * <p>
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * <p>
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.kafka.streams.kstream.internals;
+
+import org.junit.Test;
+
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertTrue;
+
+public class SessionWindowTest {
+
+    private long start = 50;
+    private long end = 100;
+    private final SessionWindow window = new SessionWindow(start, end);
+    private final TimeWindow timeWindow = new TimeWindow(start, end);
+
+    @Test
+    public void shouldNotOverlapIfOtherWindowIsBeforeThisWindow() {
+        /*
+         * This:        [-------]
+         * Other: [---]
+         */
+        assertFalse(window.overlap(new SessionWindow(0, 25)));
+        assertFalse(window.overlap(new SessionWindow(0, start - 1)));
+        assertFalse(window.overlap(new SessionWindow(start - 1, start - 1)));
+    }
+
+    @Test
+    public void shouldOverlapIfOtherWindowEndIsWithinThisWindow() {
+        /*
+         * This:        [-------]
+         * Other: [---------]
+         */
+        assertTrue(window.overlap(new SessionWindow(0, start)));
+        assertTrue(window.overlap(new SessionWindow(0, start + 1)));
+        assertTrue(window.overlap(new SessionWindow(0, 75)));
+        assertTrue(window.overlap(new SessionWindow(0, end - 1)));
+        assertTrue(window.overlap(new SessionWindow(0, end)));
+
+        assertTrue(window.overlap(new SessionWindow(start - 1, start)));
+        assertTrue(window.overlap(new SessionWindow(start - 1, start + 1)));
+        assertTrue(window.overlap(new SessionWindow(start - 1, 75)));
+        assertTrue(window.overlap(new SessionWindow(start - 1, end - 1)));
+        assertTrue(window.overlap(new SessionWindow(start - 1, end)));
+    }
+
+    @Test
+    public void shouldOverlapIfOtherWindowContainsThisWindow() {
+        /*
+         * This:        [-------]
+         * Other: [------------------]
+         */
+        assertTrue(window.overlap(new SessionWindow(0, end)));
+        assertTrue(window.overlap(new SessionWindow(0, end + 1)));
+        assertTrue(window.overlap(new SessionWindow(0, 150)));
+
+        assertTrue(window.overlap(new SessionWindow(start - 1, end)));
+        assertTrue(window.overlap(new SessionWindow(start - 1, end + 1)));
+        assertTrue(window.overlap(new SessionWindow(start - 1, 150)));
+
+        assertTrue(window.overlap(new SessionWindow(start, end)));
+        assertTrue(window.overlap(new SessionWindow(start, end + 1)));
+        assertTrue(window.overlap(new SessionWindow(start, 150)));
+    }
+
+    @Test
+    public void shouldOverlapIfOtherWindowIsWithinThisWindow() {
+        /*
+         * This:        [-------]
+         * Other:         [---]
+         */
+        assertTrue(window.overlap(new SessionWindow(start, start)));
+        assertTrue(window.overlap(new SessionWindow(start, 75)));
+        assertTrue(window.overlap(new SessionWindow(start, end)));
+        assertTrue(window.overlap(new SessionWindow(75, end)));
+        assertTrue(window.overlap(new SessionWindow(end, end)));
+    }
+
+    @Test
+    public void shouldOverlapIfOtherWindowStartIsWithinThisWindow() {
+        /*
+         * This:        [-------]
+         * Other:           [-------]
+         */
+        assertTrue(window.overlap(new SessionWindow(start, end + 1)));
+        assertTrue(window.overlap(new SessionWindow(start, 150)));
+        assertTrue(window.overlap(new SessionWindow(75, end + 1)));
+        assertTrue(window.overlap(new SessionWindow(75, 150)));
+        assertTrue(window.overlap(new SessionWindow(end, end + 1)));
+        assertTrue(window.overlap(new SessionWindow(end, 150)));
+    }
+
+    @Test
+    public void shouldNotOverlapIsOtherWindowIsAfterThisWindow() {
+        /*
+         * This:        [-------]
+         * Other:                  [---]
+         */
+        assertFalse(window.overlap(new SessionWindow(end + 1, end + 1)));
+        assertFalse(window.overlap(new SessionWindow(end + 1, 150)));
+        assertFalse(window.overlap(new SessionWindow(125, 150)));
+    }
+
+    @Test(expected = IllegalArgumentException.class)
+    public void cannotCompareSessionWindowWithDifferentWindowType() {
+        window.overlap(timeWindow);
+    }
+}
\ No newline at end of file

http://git-wip-us.apache.org/repos/asf/kafka/blob/3c605116/streams/src/test/java/org/apache/kafka/streams/state/internals/CachingSessionStoreTest.java
----------------------------------------------------------------------
diff --git a/streams/src/test/java/org/apache/kafka/streams/state/internals/CachingSessionStoreTest.java b/streams/src/test/java/org/apache/kafka/streams/state/internals/CachingSessionStoreTest.java
index a4e8df3..c603aa0 100644
--- a/streams/src/test/java/org/apache/kafka/streams/state/internals/CachingSessionStoreTest.java
+++ b/streams/src/test/java/org/apache/kafka/streams/state/internals/CachingSessionStoreTest.java
@@ -24,7 +24,7 @@ import org.apache.kafka.streams.KeyValue;
 import org.apache.kafka.streams.errors.InvalidStateStoreException;
 import org.apache.kafka.streams.kstream.Windowed;
 import org.apache.kafka.streams.kstream.internals.SessionKeySerde;
-import org.apache.kafka.streams.kstream.internals.TimeWindow;
+import org.apache.kafka.streams.kstream.internals.SessionWindow;
 import org.apache.kafka.streams.processor.internals.MockStreamsMetrics;
 import org.apache.kafka.streams.processor.internals.ProcessorRecordContext;
 import org.apache.kafka.streams.processor.internals.RecordCollector;
@@ -70,15 +70,15 @@ public class CachingSessionStoreTest {
 
     @Test
     public void shouldPutFetchFromCache() throws Exception {
-        cachingStore.put(new Windowed<>("a", new TimeWindow(0, 0)), 1L);
-        cachingStore.put(new Windowed<>("aa", new TimeWindow(0, 0)), 1L);
-        cachingStore.put(new Windowed<>("b", new TimeWindow(0, 0)), 1L);
+        cachingStore.put(new Windowed<>("a", new SessionWindow(0, 0)), 1L);
+        cachingStore.put(new Windowed<>("aa", new SessionWindow(0, 0)), 1L);
+        cachingStore.put(new Windowed<>("b", new SessionWindow(0, 0)), 1L);
 
         final KeyValueIterator<Windowed<String>, Long> a = cachingStore.findSessions("a", 0, 0);
         final KeyValueIterator<Windowed<String>, Long> b = cachingStore.findSessions("b", 0, 0);
 
-        assertEquals(KeyValue.pair(new Windowed<>("a", new TimeWindow(0, 0)), 1L), a.next());
-        assertEquals(KeyValue.pair(new Windowed<>("b", new TimeWindow(0, 0)), 1L), b.next());
+        assertEquals(KeyValue.pair(new Windowed<>("a", new SessionWindow(0, 0)), 1L), a.next());
+        assertEquals(KeyValue.pair(new Windowed<>("b", new SessionWindow(0, 0)), 1L), b.next());
         assertFalse(a.hasNext());
         assertFalse(b.hasNext());
         assertEquals(3, cache.size());
@@ -87,16 +87,16 @@ public class CachingSessionStoreTest {
     @Test
     public void shouldFetchAllSessionsWithSameRecordKey() throws Exception {
 
-        final List<KeyValue<Windowed<String>, Long>> expected = Arrays.asList(KeyValue.pair(new Windowed<>("a", new TimeWindow(0, 0)), 1L),
-                                                                                    KeyValue.pair(new Windowed<>("a", new TimeWindow(10, 10)), 2L),
-                                                                                    KeyValue.pair(new Windowed<>("a", new TimeWindow(100, 100)), 3L),
-                                                                                    KeyValue.pair(new Windowed<>("a", new TimeWindow(1000, 1000)), 4L));
+        final List<KeyValue<Windowed<String>, Long>> expected = Arrays.asList(KeyValue.pair(new Windowed<>("a", new SessionWindow(0, 0)), 1L),
+                                                                                    KeyValue.pair(new Windowed<>("a", new SessionWindow(10, 10)), 2L),
+                                                                                    KeyValue.pair(new Windowed<>("a", new SessionWindow(100, 100)), 3L),
+                                                                                    KeyValue.pair(new Windowed<>("a", new SessionWindow(1000, 1000)), 4L));
         for (KeyValue<Windowed<String>, Long> kv : expected) {
             cachingStore.put(kv.key, kv.value);
         }
 
         // add one that shouldn't appear in the results
-        cachingStore.put(new Windowed<>("aa", new TimeWindow(0, 0)), 5L);
+        cachingStore.put(new Windowed<>("aa", new SessionWindow(0, 0)), 5L);
 
 
         final List<KeyValue<Windowed<String>, Long>> results = toList(cachingStore.fetch("a"));
@@ -124,8 +124,8 @@ public class CachingSessionStoreTest {
 
     @Test
     public void shouldRemove() throws Exception {
-        final Windowed<String> a = new Windowed<>("a", new TimeWindow(0, 0));
-        final Windowed<String> b = new Windowed<>("b", new TimeWindow(0, 0));
+        final Windowed<String> a = new Windowed<>("a", new SessionWindow(0, 0));
+        final Windowed<String> b = new Windowed<>("b", new SessionWindow(0, 0));
         cachingStore.put(a, 2L);
         cachingStore.put(b, 2L);
         cachingStore.flush();
@@ -137,9 +137,9 @@ public class CachingSessionStoreTest {
 
     @Test
     public void shouldFetchCorrectlyAcrossSegments() throws Exception {
-        final Windowed<String> a1 = new Windowed<>("a", new TimeWindow(0, 0));
-        final Windowed<String> a2 = new Windowed<>("a", new TimeWindow(Segments.MIN_SEGMENT_INTERVAL, Segments.MIN_SEGMENT_INTERVAL));
-        final Windowed<String> a3 = new Windowed<>("a", new TimeWindow(Segments.MIN_SEGMENT_INTERVAL * 2, Segments.MIN_SEGMENT_INTERVAL * 2));
+        final Windowed<String> a1 = new Windowed<>("a", new SessionWindow(0, 0));
+        final Windowed<String> a2 = new Windowed<>("a", new SessionWindow(Segments.MIN_SEGMENT_INTERVAL, Segments.MIN_SEGMENT_INTERVAL));
+        final Windowed<String> a3 = new Windowed<>("a", new SessionWindow(Segments.MIN_SEGMENT_INTERVAL * 2, Segments.MIN_SEGMENT_INTERVAL * 2));
         cachingStore.put(a1, 1L);
         cachingStore.put(a2, 2L);
         cachingStore.put(a3, 3L);
@@ -153,7 +153,7 @@ public class CachingSessionStoreTest {
 
     @Test
     public void shouldClearNamespaceCacheOnClose() throws Exception {
-        final Windowed<String> a1 = new Windowed<>("a", new TimeWindow(0, 0));
+        final Windowed<String> a1 = new Windowed<>("a", new SessionWindow(0, 0));
         cachingStore.put(a1, 1L);
         assertEquals(1, cache.size());
         cachingStore.close();
@@ -175,13 +175,13 @@ public class CachingSessionStoreTest {
     @Test(expected = InvalidStateStoreException.class)
     public void shouldThrowIfTryingToRemoveFromClosedCachingStore() throws Exception {
         cachingStore.close();
-        cachingStore.remove(new Windowed<>("a", new TimeWindow(0, 0)));
+        cachingStore.remove(new Windowed<>("a", new SessionWindow(0, 0)));
     }
 
     @Test(expected = InvalidStateStoreException.class)
     public void shouldThrowIfTryingToPutIntoClosedCachingStore() throws Exception {
         cachingStore.close();
-        cachingStore.put(new Windowed<>("a", new TimeWindow(0, 0)), 1L);
+        cachingStore.put(new Windowed<>("a", new SessionWindow(0, 0)), 1L);
     }
 
     private List<KeyValue<Windowed<String>, Long>> addSessionsUntilOverflow(final String...sessionIds) {
@@ -196,11 +196,11 @@ public class CachingSessionStoreTest {
 
     private void addSingleSession(final String sessionId, final List<KeyValue<Windowed<String>, Long>> allSessions) {
         final int timestamp = allSessions.size() * 10;
-        final Windowed<String> key = new Windowed<>(sessionId, new TimeWindow(timestamp, timestamp));
+        final Windowed<String> key = new Windowed<>(sessionId, new SessionWindow(timestamp, timestamp));
         final Long value = 1L;
         cachingStore.put(key, value);
         allSessions.add(KeyValue.pair(key, value));
     }
 
 
-}
\ No newline at end of file
+}

http://git-wip-us.apache.org/repos/asf/kafka/blob/3c605116/streams/src/test/java/org/apache/kafka/streams/state/internals/RocksDBSegmentedBytesStoreTest.java
----------------------------------------------------------------------
diff --git a/streams/src/test/java/org/apache/kafka/streams/state/internals/RocksDBSegmentedBytesStoreTest.java b/streams/src/test/java/org/apache/kafka/streams/state/internals/RocksDBSegmentedBytesStoreTest.java
index 9ff2762..7fe490c 100644
--- a/streams/src/test/java/org/apache/kafka/streams/state/internals/RocksDBSegmentedBytesStoreTest.java
+++ b/streams/src/test/java/org/apache/kafka/streams/state/internals/RocksDBSegmentedBytesStoreTest.java
@@ -24,7 +24,7 @@ import org.apache.kafka.common.utils.Utils;
 import org.apache.kafka.streams.KeyValue;
 import org.apache.kafka.streams.kstream.Windowed;
 import org.apache.kafka.streams.kstream.internals.SessionKeySerde;
-import org.apache.kafka.streams.kstream.internals.TimeWindow;
+import org.apache.kafka.streams.kstream.internals.SessionWindow;
 import org.apache.kafka.streams.processor.internals.MockStreamsMetrics;
 import org.apache.kafka.streams.state.KeyValueIterator;
 import org.apache.kafka.test.MockProcessorContext;
@@ -79,13 +79,13 @@ public class RocksDBSegmentedBytesStoreTest {
     @Test
     public void shouldPutAndFetch() throws Exception {
         final String key = "a";
-        bytesStore.put(serializeKey(new Windowed<>(key, new TimeWindow(10, 10L))), serializeValue(10L));
-        bytesStore.put(serializeKey(new Windowed<>(key, new TimeWindow(500L, 1000L))), serializeValue(50L));
-        bytesStore.put(serializeKey(new Windowed<>(key, new TimeWindow(1500L, 2000L))), serializeValue(100L));
-        bytesStore.put(serializeKey(new Windowed<>(key, new TimeWindow(2500L, 3000L))), serializeValue(200L));
+        bytesStore.put(serializeKey(new Windowed<>(key, new SessionWindow(10, 10L))), serializeValue(10L));
+        bytesStore.put(serializeKey(new Windowed<>(key, new SessionWindow(500L, 1000L))), serializeValue(50L));
+        bytesStore.put(serializeKey(new Windowed<>(key, new SessionWindow(1500L, 2000L))), serializeValue(100L));
+        bytesStore.put(serializeKey(new Windowed<>(key, new SessionWindow(2500L, 3000L))), serializeValue(200L));
 
-        final List<KeyValue<Windowed<String>, Long>> expected = Arrays.asList(KeyValue.pair(new Windowed<>(key, new TimeWindow(10, 10)), 10L),
-                                                                                    KeyValue.pair(new Windowed<>(key, new TimeWindow(500, 1000)), 50L));
+        final List<KeyValue<Windowed<String>, Long>> expected = Arrays.asList(KeyValue.pair(new Windowed<>(key, new SessionWindow(10, 10)), 10L),
+                                                                                    KeyValue.pair(new Windowed<>(key, new SessionWindow(500, 1000)), 50L));
 
         final KeyValueIterator<Bytes, byte[]> values = bytesStore.fetch(Bytes.wrap(key.getBytes()), 0, 1000L);
         assertEquals(expected, toList(values));
@@ -94,18 +94,18 @@ public class RocksDBSegmentedBytesStoreTest {
     @Test
     public void shouldFindValuesWithinRange() throws Exception {
         final String key = "a";
-        bytesStore.put(serializeKey(new Windowed<>(key, new TimeWindow(0L, 0L))), serializeValue(50L));
-        bytesStore.put(serializeKey(new Windowed<>(key, new TimeWindow(1000L, 1000L))), serializeValue(10L));
+        bytesStore.put(serializeKey(new Windowed<>(key, new SessionWindow(0L, 0L))), serializeValue(50L));
+        bytesStore.put(serializeKey(new Windowed<>(key, new SessionWindow(1000L, 1000L))), serializeValue(10L));
         final KeyValueIterator<Bytes, byte[]> results = bytesStore.fetch(Bytes.wrap(key.getBytes()), 1L, 1999L);
-        assertEquals(Collections.singletonList(KeyValue.pair(new Windowed<>(key, new TimeWindow(1000L, 1000L)), 10L)), toList(results));
+        assertEquals(Collections.singletonList(KeyValue.pair(new Windowed<>(key, new SessionWindow(1000L, 1000L)), 10L)), toList(results));
     }
 
     @Test
     public void shouldRemove() throws Exception {
-        bytesStore.put(serializeKey(new Windowed<>("a", new TimeWindow(0, 1000))), serializeValue(30L));
-        bytesStore.put(serializeKey(new Windowed<>("a", new TimeWindow(1500, 2500))), serializeValue(50L));
+        bytesStore.put(serializeKey(new Windowed<>("a", new SessionWindow(0, 1000))), serializeValue(30L));
+        bytesStore.put(serializeKey(new Windowed<>("a", new SessionWindow(1500, 2500))), serializeValue(50L));
 
-        bytesStore.remove(serializeKey(new Windowed<>("a", new TimeWindow(0, 1000))));
+        bytesStore.remove(serializeKey(new Windowed<>("a", new SessionWindow(0, 1000))));
         final KeyValueIterator<Bytes, byte[]> value = bytesStore.fetch(Bytes.wrap("a".getBytes()), 0, 1000L);
         assertFalse(value.hasNext());
     }
@@ -115,32 +115,32 @@ public class RocksDBSegmentedBytesStoreTest {
         // just to validate directories
         final Segments segments = new Segments(storeName, retention, numSegments);
         final String key = "a";
-        bytesStore.put(serializeKey(new Windowed<>(key, new TimeWindow(0L, 0L))), serializeValue(50L));
+        bytesStore.put(serializeKey(new Windowed<>(key, new SessionWindow(0L, 0L))), serializeValue(50L));
         assertEquals(Collections.singleton(segments.segmentName(0)), segmentDirs());
 
-        bytesStore.put(serializeKey(new Windowed<>(key, new TimeWindow(30000L, 60000L))), serializeValue(100L));
+        bytesStore.put(serializeKey(new Windowed<>(key, new SessionWindow(30000L, 60000L))), serializeValue(100L));
         assertEquals(Utils.mkSet(segments.segmentName(0),
                                  segments.segmentName(1)), segmentDirs());
 
-        bytesStore.put(serializeKey(new Windowed<>(key, new TimeWindow(61000L, 120000L))), serializeValue(200L));
+        bytesStore.put(serializeKey(new Windowed<>(key, new SessionWindow(61000L, 120000L))), serializeValue(200L));
         assertEquals(Utils.mkSet(segments.segmentName(0),
                                  segments.segmentName(1),
                                  segments.segmentName(2)), segmentDirs());
 
-        bytesStore.put(serializeKey(new Windowed<>(key, new TimeWindow(121000L, 180000L))), serializeValue(300L));
+        bytesStore.put(serializeKey(new Windowed<>(key, new SessionWindow(121000L, 180000L))), serializeValue(300L));
         assertEquals(Utils.mkSet(segments.segmentName(1),
                                  segments.segmentName(2),
                                  segments.segmentName(3)), segmentDirs());
 
-        bytesStore.put(serializeKey(new Windowed<>(key, new TimeWindow(181000L, 240000L))), serializeValue(400L));
+        bytesStore.put(serializeKey(new Windowed<>(key, new SessionWindow(181000L, 240000L))), serializeValue(400L));
         assertEquals(Utils.mkSet(segments.segmentName(2),
                                  segments.segmentName(3),
                                  segments.segmentName(4)), segmentDirs());
 
         final List<KeyValue<Windowed<String>, Long>> results = toList(bytesStore.fetch(Bytes.wrap(key.getBytes()), 0, 240000));
-        assertEquals(Arrays.asList(KeyValue.pair(new Windowed<>(key, new TimeWindow(61000L, 120000L)), 200L),
-                                   KeyValue.pair(new Windowed<>(key, new TimeWindow(121000L, 180000L)), 300L),
-                                   KeyValue.pair(new Windowed<>(key, new TimeWindow(181000L, 240000L)), 400L)
+        assertEquals(Arrays.asList(KeyValue.pair(new Windowed<>(key, new SessionWindow(61000L, 120000L)), 200L),
+                                   KeyValue.pair(new Windowed<>(key, new SessionWindow(121000L, 180000L)), 300L),
+                                   KeyValue.pair(new Windowed<>(key, new SessionWindow(181000L, 240000L)), 400L)
                                                  ), results);
 
     }
@@ -170,4 +170,4 @@ public class RocksDBSegmentedBytesStoreTest {
         return results;
     }
 
-}
\ No newline at end of file
+}

http://git-wip-us.apache.org/repos/asf/kafka/blob/3c605116/streams/src/test/java/org/apache/kafka/streams/state/internals/RocksDBSessionStoreTest.java
----------------------------------------------------------------------
diff --git a/streams/src/test/java/org/apache/kafka/streams/state/internals/RocksDBSessionStoreTest.java b/streams/src/test/java/org/apache/kafka/streams/state/internals/RocksDBSessionStoreTest.java
index ab4f5da..5a23a1c 100644
--- a/streams/src/test/java/org/apache/kafka/streams/state/internals/RocksDBSessionStoreTest.java
+++ b/streams/src/test/java/org/apache/kafka/streams/state/internals/RocksDBSessionStoreTest.java
@@ -21,7 +21,7 @@ import org.apache.kafka.common.metrics.Metrics;
 import org.apache.kafka.common.serialization.Serdes;
 import org.apache.kafka.streams.KeyValue;
 import org.apache.kafka.streams.kstream.Windowed;
-import org.apache.kafka.streams.kstream.internals.TimeWindow;
+import org.apache.kafka.streams.kstream.internals.SessionWindow;
 import org.apache.kafka.streams.processor.internals.MockStreamsMetrics;
 import org.apache.kafka.streams.state.KeyValueIterator;
 import org.apache.kafka.streams.state.SessionStore;
@@ -70,12 +70,12 @@ public class RocksDBSessionStoreTest {
     @Test
     public void shouldPutAndFindSessionsInRange() throws Exception {
         final String key = "a";
-        final Windowed<String> a1 = new Windowed<>(key, new TimeWindow(10, 10L));
-        final Windowed<String> a2 = new Windowed<>(key, new TimeWindow(500L, 1000L));
+        final Windowed<String> a1 = new Windowed<>(key, new SessionWindow(10, 10L));
+        final Windowed<String> a2 = new Windowed<>(key, new SessionWindow(500L, 1000L));
         sessionStore.put(a1, 1L);
         sessionStore.put(a2, 2L);
-        sessionStore.put(new Windowed<>(key, new TimeWindow(1500L, 2000L)), 1L);
-        sessionStore.put(new Windowed<>(key, new TimeWindow(2500L, 3000L)), 2L);
+        sessionStore.put(new Windowed<>(key, new SessionWindow(1500L, 2000L)), 1L);
+        sessionStore.put(new Windowed<>(key, new SessionWindow(2500L, 3000L)), 2L);
 
         final List<KeyValue<Windowed<String>, Long>> expected
                 = Arrays.asList(KeyValue.pair(a1, 1L), KeyValue.pair(a2, 2L));
@@ -87,16 +87,16 @@ public class RocksDBSessionStoreTest {
     @Test
     public void shouldFetchAllSessionsWithSameRecordKey() throws Exception {
 
-        final List<KeyValue<Windowed<String>, Long>> expected = Arrays.asList(KeyValue.pair(new Windowed<>("a", new TimeWindow(0, 0)), 1L),
-                                                                                    KeyValue.pair(new Windowed<>("a", new TimeWindow(10, 10)), 2L),
-                                                                                    KeyValue.pair(new Windowed<>("a", new TimeWindow(100, 100)), 3L),
-                                                                                    KeyValue.pair(new Windowed<>("a", new TimeWindow(1000, 1000)), 4L));
+        final List<KeyValue<Windowed<String>, Long>> expected = Arrays.asList(KeyValue.pair(new Windowed<>("a", new SessionWindow(0, 0)), 1L),
+                                                                                    KeyValue.pair(new Windowed<>("a", new SessionWindow(10, 10)), 2L),
+                                                                                    KeyValue.pair(new Windowed<>("a", new SessionWindow(100, 100)), 3L),
+                                                                                    KeyValue.pair(new Windowed<>("a", new SessionWindow(1000, 1000)), 4L));
         for (KeyValue<Windowed<String>, Long> kv : expected) {
             sessionStore.put(kv.key, kv.value);
         }
 
         // add one that shouldn't appear in the results
-        sessionStore.put(new Windowed<>("aa", new TimeWindow(0, 0)), 5L);
+        sessionStore.put(new Windowed<>("aa", new SessionWindow(0, 0)), 5L);
 
         final List<KeyValue<Windowed<String>, Long>> results = toList(sessionStore.fetch("a"));
         assertEquals(expected, results);
@@ -107,22 +107,22 @@ public class RocksDBSessionStoreTest {
     @Test
     public void shouldFindValuesWithinMergingSessionWindowRange() throws Exception {
         final String key = "a";
-        sessionStore.put(new Windowed<>(key, new TimeWindow(0L, 0L)), 1L);
-        sessionStore.put(new Windowed<>(key, new TimeWindow(1000L, 1000L)), 2L);
+        sessionStore.put(new Windowed<>(key, new SessionWindow(0L, 0L)), 1L);
+        sessionStore.put(new Windowed<>(key, new SessionWindow(1000L, 1000L)), 2L);
         final KeyValueIterator<Windowed<String>, Long> results = sessionStore.findSessions(key, -1, 1000L);
 
         final List<KeyValue<Windowed<String>, Long>> expected = Arrays.asList(
-                KeyValue.pair(new Windowed<>(key, new TimeWindow(0L, 0L)), 1L),
-                KeyValue.pair(new Windowed<>(key, new TimeWindow(1000L, 1000L)), 2L));
+                KeyValue.pair(new Windowed<>(key, new SessionWindow(0L, 0L)), 1L),
+                KeyValue.pair(new Windowed<>(key, new SessionWindow(1000L, 1000L)), 2L));
         assertEquals(expected, toList(results));
     }
 
     @Test
     public void shouldRemove() throws Exception {
-        sessionStore.put(new Windowed<>("a", new TimeWindow(0, 1000)), 1L);
-        sessionStore.put(new Windowed<>("a", new TimeWindow(1500, 2500)), 2L);
+        sessionStore.put(new Windowed<>("a", new SessionWindow(0, 1000)), 1L);
+        sessionStore.put(new Windowed<>("a", new SessionWindow(1500, 2500)), 2L);
 
-        sessionStore.remove(new Windowed<>("a", new TimeWindow(0, 1000)));
+        sessionStore.remove(new Windowed<>("a", new SessionWindow(0, 1000)));
         assertFalse(sessionStore.findSessions("a", 0, 1000L).hasNext());
 
         assertTrue(sessionStore.findSessions("a", 1500, 2500).hasNext());
@@ -130,11 +130,11 @@ public class RocksDBSessionStoreTest {
 
     @Test
     public void shouldFindSessionsToMerge() throws Exception {
-        final Windowed<String> session1 = new Windowed<>("a", new TimeWindow(0, 100));
-        final Windowed<String> session2 = new Windowed<>("a", new TimeWindow(101, 200));
-        final Windowed<String> session3 = new Windowed<>("a", new TimeWindow(201, 300));
-        final Windowed<String> session4 = new Windowed<>("a", new TimeWindow(301, 400));
-        final Windowed<String> session5 = new Windowed<>("a", new TimeWindow(401, 500));
+        final Windowed<String> session1 = new Windowed<>("a", new SessionWindow(0, 100));
+        final Windowed<String> session2 = new Windowed<>("a", new SessionWindow(101, 200));
+        final Windowed<String> session3 = new Windowed<>("a", new SessionWindow(201, 300));
+        final Windowed<String> session4 = new Windowed<>("a", new SessionWindow(301, 400));
+        final Windowed<String> session5 = new Windowed<>("a", new SessionWindow(401, 500));
         sessionStore.put(session1, 1L);
         sessionStore.put(session2, 2L);
         sessionStore.put(session3, 3L);
@@ -155,4 +155,4 @@ public class RocksDBSessionStoreTest {
     }
 
 
-}
\ No newline at end of file
+}


Mime
View raw message