{"action":"created","comment":{"url":"https://api.github.com/repos/apache/kafka/pulls/comments/92409491","pull_request_review_id":12923581,"id":92409491,"diff_hunk":"@@ -0,0 +1,196 @@\n+/**\n+ * Licensed to the Apache Software Foundation (ASF) under one or more\n+ * contributor license agreements.  See the NOTICE file distributed with\n+ * this work for additional information regarding copyright ownership.\n+ * The ASF licenses this file to You under the Apache License, Version 2.0\n+ * (the \"License\"); you may not use this file except in compliance with\n+ * the License.  You may obtain a copy of the License at\n+ * <p>\n+ * http://www.apache.org/licenses/LICENSE-2.0\n+ * <p>\n+ * Unless required by applicable law or agreed to in writing, software\n+ * distributed under the License is distributed on an \"AS IS\" BASIS,\n+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.\n+ * See the License for the specific language governing permissions and\n+ * limitations under the License.\n+ */\n+package org.apache.kafka.streams.state.internals;\n+\n+import org.apache.kafka.common.serialization.Serde;\n+import org.apache.kafka.common.serialization.Serdes;\n+import org.apache.kafka.common.utils.Bytes;\n+import org.apache.kafka.streams.KeyValue;\n+import org.apache.kafka.streams.kstream.Windowed;\n+import org.apache.kafka.streams.kstream.internals.SessionKeyBinaryConverter;\n+import org.apache.kafka.streams.kstream.internals.TimeWindow;\n+import org.apache.kafka.streams.processor.ProcessorContext;\n+import org.apache.kafka.streams.processor.StateStore;\n+import org.apache.kafka.streams.state.KeyValueIterator;\n+import org.apache.kafka.streams.state.SessionStore;\n+import org.apache.kafka.streams.state.StateSerdes;\n+\n+import java.util.List;\n+\n+class RocksDBSessionStore<K, AGG> implements SessionStore<K, AGG> {\n+\n+    private final Serde<K> keySerde;\n+    private final Serde<AGG> aggSerde;\n+    private final SegmentedBytesStore bytesStore;\n+    private StateSerdes<K, AGG> serdes;\n+\n+\n+    RocksDBSessionStore(final SegmentedBytesStore bytesStore,\n+                        final Serde<K> keySerde,\n+                        final Serde<AGG> aggSerde) {\n+        this.keySerde = keySerde;\n+        this.bytesStore = bytesStore;\n+        this.aggSerde = aggSerde;\n+    }\n+\n+\n+    @SuppressWarnings(\"unchecked\")\n+    @Override\n+    public KeyValueIterator<Windowed<K>, AGG> findSessionsToMerge(final K key, final long earliestSessionEndTime, final long latestSessionStartTime) {\n+        final KeyValueIterator<Bytes, byte[]> bytesIterator = bytesStore.fetch(Bytes.wrap(serdes.rawKey(key)), earliestSessionEndTime, latestSessionStartTime);\n+        return new SessionStoreIterator(bytesIterator, serdes);\n+    }\n+\n+\n+    @Override\n+    public void remove(final Windowed<K> key) {\n+        bytesStore.remove(SessionKeyBinaryConverter.toBinary(key, serdes.keySerializer()));\n+    }\n+\n+    @Override\n+    public void put(final Windowed<K> sessionKey, final AGG aggregate) {\n+        bytesStore.put(SessionKeyBinaryConverter.toBinary(sessionKey, serdes.keySerializer()), aggSerde.serializer().serialize(bytesStore.name(), aggregate));\n+    }\n+\n+    @Override\n+    public String name() {\n+        return bytesStore.name();\n+    }\n+\n+    @Override\n+    @SuppressWarnings(\"unchecked\")\n+    public void init(final ProcessorContext context, final StateStore root) {\n+        this.serdes = new StateSerdes<>(bytesStore.name(),\n+                                        keySerde == null ? (Serde<K>) context.keySerde() : keySerde,\n+                                        aggSerde == null ? (Serde<AGG>) context.valueSerde() : aggSerde);\n+\n+        bytesStore.init(context, root);\n+    }\n+\n+    @Override\n+    public void flush() {\n+        bytesStore.flush();\n+    }\n+\n+    @Override\n+    public void close() {\n+        bytesStore.close();\n+    }\n+\n+    @Override\n+    public boolean persistent() {\n+        return true;\n+    }\n+\n+    @Override\n+    public boolean isOpen() {\n+        return bytesStore.isOpen();\n+    }\n+\n+    @Override\n+    public KeyValueIterator<Windowed<K>, AGG> fetch(final K key) {\n+        return findSessionsToMerge(key, 0, Long.MAX_VALUE);\n+    }\n+\n+    static class SessionKeySchema implements RocksDBSegmentedBytesStore.KeySchema {\n+\n+        @Override\n+        public Bytes upperRange(final Bytes key, final long to) {\n+            final Windowed<Bytes> sessionKey = new Windowed<>(key, new TimeWindow(to, Long.MAX_VALUE));\n+            return SessionKeyBinaryConverter.toBinary(sessionKey, Serdes.Bytes().serializer());\n+        }\n+\n+        @Override\n+        public Bytes lowerRange(final Bytes key, final long from) {\n+            final Windowed<Bytes> sessionKey = new Windowed<>(key, new TimeWindow(0, Math.max(0, from)));\n+            return SessionKeyBinaryConverter.toBinary(sessionKey, Serdes.Bytes().serializer());\n+        }\n+\n+        @Override\n+        public long segmentTimestamp(final Bytes key) {\n+            return SessionKeyBinaryConverter.extractEnd(key.get());\n+        }\n+\n+        @Override\n+        public HasNextCondition<Bytes> hasNextCondition(final Bytes binaryKey, final long from, final long to) {\n+            return hasNextMatchingSession(from, to, binaryKey);\n+        }\n+\n+        @Override\n+        public List<Segment> segmentsToSearch(final Segments segments, final long from, final long to) {\n+            return segments.segments(from, Long.MAX_VALUE);\n+        }\n+\n+    }\n+\n+    static HasNextCondition<Bytes> hasNextMatchingSession(final long earliestEndTime, final long latestStartTime, final Bytes binaryKey) {\n+        return new HasNextCondition<Bytes>() {\n+            @Override\n+            public boolean hasNext(final KeyValueIterator<Bytes, ?> iterator) {\n+                if (iterator.hasNext()) {\n+                    final Bytes bytes = iterator.peekNextKey();\n+                    final Bytes keyBytes = Bytes.wrap(SessionKeyBinaryConverter.extractKeyBytes(bytes.get()));\n+                    if (!keyBytes.equals(binaryKey)) {\n+                        return false;\n+                    }\n+                    final long start = SessionKeyBinaryConverter.extractStart(bytes.get());\n+                    final long end = SessionKeyBinaryConverter.extractEnd(bytes.get());\n+                    return end >= earliestEndTime && start <= latestStartTime;\n+                }\n+                return false;\n+            }\n+        };\n+    }\n+\n+    static class SessionStoreIterator<K, AGG> implements KeyValueIterator<Windowed<K>, AGG> {","path":"streams/src/main/java/org/apache/kafka/streams/state/internals/RocksDBSessionStore.java","position":159,"original_position":159,"commit_id":"090bace37f06f2ae5ff238069f8755d508bdfb54","original_commit_id":"090bace37f06f2ae5ff238069f8755d508bdfb54","user":{"login":"dguy","id":54829,"avatar_url":"https://avatars.githubusercontent.com/u/54829?v=3","gravatar_id":"","url":"https://api.github.com/users/dguy","html_url":"https://github.com/dguy","followers_url":"https://api.github.com/users/dguy/followers","following_url":"https://api.github.com/users/dguy/following{/other_user}","gists_url":"https://api.github.com/users/dguy/gists{/gist_id}","starred_url":"https://api.github.com/users/dguy/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/dguy/subscriptions","organizations_url":"https://api.github.com/users/dguy/orgs","repos_url":"https://api.github.com/users/dguy/repos","events_url":"https://api.github.com/users/dguy/events{/privacy}","received_events_url":"https://api.github.com/users/dguy/received_events","type":"User","site_admin":false},"body":"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.","created_at":"2016-12-14T14:58:04Z","updated_at":"2016-12-14T14:58:04Z","html_url":"https://github.com/apache/kafka/pull/2166#discussion_r92409491","pull_request_url":"https://api.github.com/repos/apache/kafka/pulls/2166","_links":{"self":{"href":"https://api.github.com/repos/apache/kafka/pulls/comments/92409491"},"html":{"href":"https://github.com/apache/kafka/pull/2166#discussion_r92409491"},"pull_request":{"href":"https://api.github.com/repos/apache/kafka/pulls/2166"}}},"pull_request":{"url":"https://api.github.com/repos/apache/kafka/pulls/2166","id":95208371,"html_url":"https://github.com/apache/kafka/pull/2166","diff_url":"https://github.com/apache/kafka/pull/2166.diff","patch_url":"https://github.com/apache/kafka/pull/2166.patch","issue_url":"https://api.github.com/repos/apache/kafka/issues/2166","number":2166,"state":"open","locked":false,"title":"KAFKA-3452: Support session windows","user":{"login":"dguy","id":54829,"avatar_url":"https://avatars.githubusercontent.com/u/54829?v=3","gravatar_id":"","url":"https://api.github.com/users/dguy","html_url":"https://github.com/dguy","followers_url":"https://api.github.com/users/dguy/followers","following_url":"https://api.github.com/users/dguy/following{/other_user}","gists_url":"https://api.github.com/users/dguy/gists{/gist_id}","starred_url":"https://api.github.com/users/dguy/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/dguy/subscriptions","organizations_url":"https://api.github.com/users/dguy/orgs","repos_url":"https://api.github.com/users/dguy/repos","events_url":"https://api.github.com/users/dguy/events{/privacy}","received_events_url":"https://api.github.com/users/dguy/received_events","type":"User","site_admin":false},"body":"Add support for SessionWindows based on design detailed in https://cwiki.apache.org/confluence/display/KAFKA/KIP-94+Session+Windows.\r\nThis includes refactoring of the RocksDBWindowStore such that functionality common with the RocksDBSessionStore isn't duplicated.","created_at":"2016-11-24T16:48:23Z","updated_at":"2016-12-14T14:58:04Z","closed_at":null,"merged_at":null,"merge_commit_sha":"13d0069a9cd1c41ac451f0f4b4917998fd17afa1","assignee":null,"assignees":[],"milestone":null,"commits_url":"https://api.github.com/repos/apache/kafka/pulls/2166/commits","review_comments_url":"https://api.github.com/repos/apache/kafka/pulls/2166/comments","review_comment_url":"https://api.github.com/repos/apache/kafka/pulls/comments{/number}","comments_url":"https://api.github.com/repos/apache/kafka/issues/2166/comments","statuses_url":"https://api.github.com/repos/apache/kafka/statuses/090bace37f06f2ae5ff238069f8755d508bdfb54","head":{"label":"dguy:kafka-3452-session-merge","ref":"kafka-3452-session-merge","sha":"090bace37f06f2ae5ff238069f8755d508bdfb54","user":{"login":"dguy","id":54829,"avatar_url":"https://avatars.githubusercontent.com/u/54829?v=3","gravatar_id":"","url":"https://api.github.com/users/dguy","html_url":"https://github.com/dguy","followers_url":"https://api.github.com/users/dguy/followers","following_url":"https://api.github.com/users/dguy/following{/other_user}","gists_url":"https://api.github.com/users/dguy/gists{/gist_id}","starred_url":"https://api.github.com/users/dguy/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/dguy/subscriptions","organizations_url":"https://api.github.com/users/dguy/orgs","repos_url":"https://api.github.com/users/dguy/repos","events_url":"https://api.github.com/users/dguy/events{/privacy}","received_events_url":"https://api.github.com/users/dguy/received_events","type":"User","site_admin":false},"repo":{"id":56393114,"name":"kafka","full_name":"dguy/kafka","owner":{"login":"dguy","id":54829,"avatar_url":"https://avatars.githubusercontent.com/u/54829?v=3","gravatar_id":"","url":"https://api.github.com/users/dguy","html_url":"https://github.com/dguy","followers_url":"https://api.github.com/users/dguy/followers","following_url":"https://api.github.com/users/dguy/following{/other_user}","gists_url":"https://api.github.com/users/dguy/gists{/gist_id}","starred_url":"https://api.github.com/users/dguy/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/dguy/subscriptions","organizations_url":"https://api.github.com/users/dguy/orgs","repos_url":"https://api.github.com/users/dguy/repos","events_url":"https://api.github.com/users/dguy/events{/privacy}","received_events_url":"https://api.github.com/users/dguy/received_events","type":"User","site_admin":false},"private":false,"html_url":"https://github.com/dguy/kafka","description":"Mirror of Apache Kafka","fork":true,"url":"https://api.github.com/repos/dguy/kafka","forks_url":"https://api.github.com/repos/dguy/kafka/forks","keys_url":"https://api.github.com/repos/dguy/kafka/keys{/key_id}","collaborators_url":"https://api.github.com/repos/dguy/kafka/collaborators{/collaborator}","teams_url":"https://api.github.com/repos/dguy/kafka/teams","hooks_url":"https://api.github.com/repos/dguy/kafka/hooks","issue_events_url":"https://api.github.com/repos/dguy/kafka/issues/events{/number}","events_url":"https://api.github.com/repos/dguy/kafka/events","assignees_url":"https://api.github.com/repos/dguy/kafka/assignees{/user}","branches_url":"https://api.github.com/repos/dguy/kafka/branches{/branch}","tags_url":"https://api.github.com/repos/dguy/kafka/tags","blobs_url":"https://api.github.com/repos/dguy/kafka/git/blobs{/sha}","git_tags_url":"https://api.github.com/repos/dguy/kafka/git/tags{/sha}","git_refs_url":"https://api.github.com/repos/dguy/kafka/git/refs{/sha}","trees_url":"https://api.github.com/repos/dguy/kafka/git/trees{/sha}","statuses_url":"https://api.github.com/repos/dguy/kafka/statuses/{sha}","languages_url":"https://api.github.com/repos/dguy/kafka/languages","stargazers_url":"https://api.github.com/repos/dguy/kafka/stargazers","contributors_url":"https://api.github.com/repos/dguy/kafka/contributors","subscribers_url":"https://api.github.com/repos/dguy/kafka/subscribers","subscription_url":"https://api.github.com/repos/dguy/kafka/subscription","commits_url":"https://api.github.com/repos/dguy/kafka/commits{/sha}","git_commits_url":"https://api.github.com/repos/dguy/kafka/git/commits{/sha}","comments_url":"https://api.github.com/repos/dguy/kafka/comments{/number}","issue_comment_url":"https://api.github.com/repos/dguy/kafka/issues/comments{/number}","contents_url":"https://api.github.com/repos/dguy/kafka/contents/{+path}","compare_url":"https://api.github.com/repos/dguy/kafka/compare/{base}...{head}","merges_url":"https://api.github.com/repos/dguy/kafka/merges","archive_url":"https://api.github.com/repos/dguy/kafka/{archive_format}{/ref}","downloads_url":"https://api.github.com/repos/dguy/kafka/downloads","issues_url":"https://api.github.com/repos/dguy/kafka/issues{/number}","pulls_url":"https://api.github.com/repos/dguy/kafka/pulls{/number}","milestones_url":"https://api.github.com/repos/dguy/kafka/milestones{/number}","notifications_url":"https://api.github.com/repos/dguy/kafka/notifications{?since,all,participating}","labels_url":"https://api.github.com/repos/dguy/kafka/labels{/name}","releases_url":"https://api.github.com/repos/dguy/kafka/releases{/id}","deployments_url":"https://api.github.com/repos/dguy/kafka/deployments","created_at":"2016-04-16T16:12:53Z","updated_at":"2016-04-16T16:12:57Z","pushed_at":"2016-12-14T14:14:29Z","git_url":"git://github.com/dguy/kafka.git","ssh_url":"git@github.com:dguy/kafka.git","clone_url":"https://github.com/dguy/kafka.git","svn_url":"https://github.com/dguy/kafka","homepage":null,"size":37680,"stargazers_count":0,"watchers_count":0,"language":"Java","has_issues":false,"has_downloads":true,"has_wiki":false,"has_pages":false,"forks_count":0,"mirror_url":null,"open_issues_count":0,"forks":0,"open_issues":0,"watchers":0,"default_branch":"trunk"}},"base":{"label":"apache:trunk","ref":"trunk","sha":"53428694a624a1956cb8fa3a6a674638f9b09816","user":{"login":"apache","id":47359,"avatar_url":"https://avatars.githubusercontent.com/u/47359?v=3","gravatar_id":"","url":"https://api.github.com/users/apache","html_url":"https://github.com/apache","followers_url":"https://api.github.com/users/apache/followers","following_url":"https://api.github.com/users/apache/following{/other_user}","gists_url":"https://api.github.com/users/apache/gists{/gist_id}","starred_url":"https://api.github.com/users/apache/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/apache/subscriptions","organizations_url":"https://api.github.com/users/apache/orgs","repos_url":"https://api.github.com/users/apache/repos","events_url":"https://api.github.com/users/apache/events{/privacy}","received_events_url":"https://api.github.com/users/apache/received_events","type":"Organization","site_admin":false},"repo":{"id":2211243,"name":"kafka","full_name":"apache/kafka","owner":{"login":"apache","id":47359,"avatar_url":"https://avatars.githubusercontent.com/u/47359?v=3","gravatar_id":"","url":"https://api.github.com/users/apache","html_url":"https://github.com/apache","followers_url":"https://api.github.com/users/apache/followers","following_url":"https://api.github.com/users/apache/following{/other_user}","gists_url":"https://api.github.com/users/apache/gists{/gist_id}","starred_url":"https://api.github.com/users/apache/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/apache/subscriptions","organizations_url":"https://api.github.com/users/apache/orgs","repos_url":"https://api.github.com/users/apache/repos","events_url":"https://api.github.com/users/apache/events{/privacy}","received_events_url":"https://api.github.com/users/apache/received_events","type":"Organization","site_admin":false},"private":false,"html_url":"https://github.com/apache/kafka","description":"Mirror of Apache Kafka","fork":false,"url":"https://api.github.com/repos/apache/kafka","forks_url":"https://api.github.com/repos/apache/kafka/forks","keys_url":"https://api.github.com/repos/apache/kafka/keys{/key_id}","collaborators_url":"https://api.github.com/repos/apache/kafka/collaborators{/collaborator}","teams_url":"https://api.github.com/repos/apache/kafka/teams","hooks_url":"https://api.github.com/repos/apache/kafka/hooks","issue_events_url":"https://api.github.com/repos/apache/kafka/issues/events{/number}","events_url":"https://api.github.com/repos/apache/kafka/events","assignees_url":"https://api.github.com/repos/apache/kafka/assignees{/user}","branches_url":"https://api.github.com/repos/apache/kafka/branches{/branch}","tags_url":"https://api.github.com/repos/apache/kafka/tags","blobs_url":"https://api.github.com/repos/apache/kafka/git/blobs{/sha}","git_tags_url":"https://api.github.com/repos/apache/kafka/git/tags{/sha}","git_refs_url":"https://api.github.com/repos/apache/kafka/git/refs{/sha}","trees_url":"https://api.github.com/repos/apache/kafka/git/trees{/sha}","statuses_url":"https://api.github.com/repos/apache/kafka/statuses/{sha}","languages_url":"https://api.github.com/repos/apache/kafka/languages","stargazers_url":"https://api.github.com/repos/apache/kafka/stargazers","contributors_url":"https://api.github.com/repos/apache/kafka/contributors","subscribers_url":"https://api.github.com/repos/apache/kafka/subscribers","subscription_url":"https://api.github.com/repos/apache/kafka/subscription","commits_url":"https://api.github.com/repos/apache/kafka/commits{/sha}","git_commits_url":"https://api.github.com/repos/apache/kafka/git/commits{/sha}","comments_url":"https://api.github.com/repos/apache/kafka/comments{/number}","issue_comment_url":"https://api.github.com/repos/apache/kafka/issues/comments{/number}","contents_url":"https://api.github.com/repos/apache/kafka/contents/{+path}","compare_url":"https://api.github.com/repos/apache/kafka/compare/{base}...{head}","merges_url":"https://api.github.com/repos/apache/kafka/merges","archive_url":"https://api.github.com/repos/apache/kafka/{archive_format}{/ref}","downloads_url":"https://api.github.com/repos/apache/kafka/downloads","issues_url":"https://api.github.com/repos/apache/kafka/issues{/number}","pulls_url":"https://api.github.com/repos/apache/kafka/pulls{/number}","milestones_url":"https://api.github.com/repos/apache/kafka/milestones{/number}","notifications_url":"https://api.github.com/repos/apache/kafka/notifications{?since,all,participating}","labels_url":"https://api.github.com/repos/apache/kafka/labels{/name}","releases_url":"https://api.github.com/repos/apache/kafka/releases{/id}","deployments_url":"https://api.github.com/repos/apache/kafka/deployments","created_at":"2011-08-15T18:06:16Z","updated_at":"2016-12-14T14:31:29Z","pushed_at":"2016-12-14T14:14:29Z","git_url":"git://github.com/apache/kafka.git","ssh_url":"git@github.com:apache/kafka.git","clone_url":"https://github.com/apache/kafka.git","svn_url":"https://github.com/apache/kafka","homepage":null,"size":37554,"stargazers_count":4273,"watchers_count":4273,"language":"Java","has_issues":false,"has_downloads":true,"has_wiki":false,"has_pages":false,"forks_count":2666,"mirror_url":"git://git.apache.org/kafka.git","open_issues_count":307,"forks":2666,"open_issues":307,"watchers":4273,"default_branch":"trunk"}},"_links":{"self":{"href":"https://api.github.com/repos/apache/kafka/pulls/2166"},"html":{"href":"https://github.com/apache/kafka/pull/2166"},"issue":{"href":"https://api.github.com/repos/apache/kafka/issues/2166"},"comments":{"href":"https://api.github.com/repos/apache/kafka/issues/2166/comments"},"review_comments":{"href":"https://api.github.com/repos/apache/kafka/pulls/2166/comments"},"review_comment":{"href":"https://api.github.com/repos/apache/kafka/pulls/comments{/number}"},"commits":{"href":"https://api.github.com/repos/apache/kafka/pulls/2166/commits"},"statuses":{"href":"https://api.github.com/repos/apache/kafka/statuses/090bace37f06f2ae5ff238069f8755d508bdfb54"}}}}
{"actor":{"display_login":"dguy"}}