{"action":"created","comment":{"url":"https://api.github.com/repos/apache/samza/pulls/comments/180913735","pull_request_review_id":111421446,"id":180913735,"diff_hunk":"@@ -0,0 +1,173 @@\n+/*\n+ * Licensed to the Apache Software Foundation (ASF) under one\n+ * or more contributor license agreements.  See the NOTICE file\n+ * distributed with this work for additional information\n+ * regarding copyright ownership.  The ASF licenses this file\n+ * to you under the Apache License, Version 2.0 (the\n+ * \"License\"); you may not use this file except in compliance\n+ * with the License.  You may obtain a copy of the License at\n+ *\n+ *   http://www.apache.org/licenses/LICENSE-2.0\n+ *\n+ * Unless required by applicable law or agreed to in writing,\n+ * software distributed under the License is distributed on an\n+ * \"AS IS\" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY\n+ * KIND, either express or implied.  See the License for the\n+ * specific language governing permissions and limitations\n+ * under the License.\n+ */\n+\n+package org.apache.samza.system.inmemory;\n+\n+import java.util.ArrayList;\n+import java.util.Collections;\n+import java.util.LinkedList;\n+import java.util.List;\n+import java.util.Map;\n+import java.util.Set;\n+import java.util.concurrent.ConcurrentHashMap;\n+import java.util.stream.Collectors;\n+import org.apache.samza.Partition;\n+import org.apache.samza.system.EndOfStreamMessage;\n+import org.apache.samza.system.IncomingMessageEnvelope;\n+import org.apache.samza.system.StreamSpec;\n+import org.apache.samza.system.SystemStream;\n+import org.apache.samza.system.SystemStreamMetadata;\n+import org.apache.samza.system.SystemStreamPartition;\n+import org.slf4j.Logger;\n+import org.slf4j.LoggerFactory;\n+\n+\n+/**\n+ * Initial draft of in-memory manager. It is test only and not meant for production use right now.\n+ */\n+public class InMemoryManager {\n+  private static final Logger LOG = LoggerFactory.getLogger(InMemoryManager.class);\n+  private static final int DEFAULT_PARTITION_COUNT = 1;\n+\n+  private final ConcurrentHashMap<SystemStreamPartition, List<IncomingMessageEnvelope>> bufferedMessages;\n+  private final Map<SystemStream, Integer> systemStreamToPartitions;\n+\n+  public InMemoryManager() {\n+    bufferedMessages = new ConcurrentHashMap<>();\n+    systemStreamToPartitions = new ConcurrentHashMap<>();\n+  }\n+\n+  private List<IncomingMessageEnvelope> newSynchronizedLinkedList() {\n+    return  Collections.synchronizedList(new LinkedList<IncomingMessageEnvelope>());\n+  }\n+\n+  public void put(SystemStreamPartition ssp, Object message) {\n+    put(ssp, null, message);\n+  }\n+\n+  public void put(SystemStreamPartition ssp, Object key, Object message) {\n+    List<IncomingMessageEnvelope> messages = bufferedMessages.get(ssp);\n+    String offset = String.valueOf(messages.size());\n+\n+    if (message instanceof EndOfStreamMessage) {\n+      offset = IncomingMessageEnvelope.END_OF_STREAM_OFFSET;\n+    }\n+\n+    IncomingMessageEnvelope messageEnvelope = new IncomingMessageEnvelope(ssp, offset, key, message);\n+    bufferedMessages.get(ssp)\n+        .add(messageEnvelope);\n+  }\n+\n+  public Map<SystemStreamPartition, List<IncomingMessageEnvelope>> poll(Map<SystemStreamPartition, String> sspsToOffsets) {\n+    return sspsToOffsets.entrySet()\n+        .stream()\n+        .collect(Collectors.toMap(\n+            Map.Entry::getKey,\n+            entry -> poll(entry.getKey(), entry.getValue())));\n+  }\n+\n+  private List<IncomingMessageEnvelope> poll(SystemStreamPartition ssp, String offset) {\n+    int startingOffset = Integer.parseInt(offset);\n+    List<IncomingMessageEnvelope> messageEnvelopesForSSP = bufferedMessages.getOrDefault(ssp, new LinkedList<>());\n+\n+    if (startingOffset >= messageEnvelopesForSSP.size()) {\n+      return new ArrayList<>();\n+    }\n+\n+    return messageEnvelopesForSSP.subList(startingOffset, messageEnvelopesForSSP.size());","path":"samza-core/src/main/java/org/apache/samza/system/inmemory/InMemoryManager.java","position":93,"original_position":93,"commit_id":"3b4068914d906200070150fb6a72a342a1ac70ed","original_commit_id":"3b4068914d906200070150fb6a72a342a1ac70ed","user":{"login":"xinyuiscool","id":24738423,"avatar_url":"https://avatars0.githubusercontent.com/u/24738423?v=4","gravatar_id":"","url":"https://api.github.com/users/xinyuiscool","html_url":"https://github.com/xinyuiscool","followers_url":"https://api.github.com/users/xinyuiscool/followers","following_url":"https://api.github.com/users/xinyuiscool/following{/other_user}","gists_url":"https://api.github.com/users/xinyuiscool/gists{/gist_id}","starred_url":"https://api.github.com/users/xinyuiscool/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/xinyuiscool/subscriptions","organizations_url":"https://api.github.com/users/xinyuiscool/orgs","repos_url":"https://api.github.com/users/xinyuiscool/repos","events_url":"https://api.github.com/users/xinyuiscool/events{/privacy}","received_events_url":"https://api.github.com/users/xinyuiscool/received_events","type":"User","site_admin":false},"body":"We do not remove the elements polled from the ssp?","created_at":"2018-04-11T22:12:39Z","updated_at":"2018-04-11T22:59:46Z","html_url":"https://github.com/apache/samza/pull/459#discussion_r180913735","pull_request_url":"https://api.github.com/repos/apache/samza/pulls/459","author_association":"CONTRIBUTOR","_links":{"self":{"href":"https://api.github.com/repos/apache/samza/pulls/comments/180913735"},"html":{"href":"https://github.com/apache/samza/pull/459#discussion_r180913735"},"pull_request":{"href":"https://api.github.com/repos/apache/samza/pulls/459"}}},"pull_request":{"url":"https://api.github.com/repos/apache/samza/pulls/459","id":178137489,"html_url":"https://github.com/apache/samza/pull/459","diff_url":"https://github.com/apache/samza/pull/459.diff","patch_url":"https://github.com/apache/samza/pull/459.patch","issue_url":"https://api.github.com/repos/apache/samza/issues/459","number":459,"state":"open","locked":false,"title":"Initial version for in memory system.","user":{"login":"bharathkk","id":1001798,"avatar_url":"https://avatars2.githubusercontent.com/u/1001798?v=4","gravatar_id":"","url":"https://api.github.com/users/bharathkk","html_url":"https://github.com/bharathkk","followers_url":"https://api.github.com/users/bharathkk/followers","following_url":"https://api.github.com/users/bharathkk/following{/other_user}","gists_url":"https://api.github.com/users/bharathkk/gists{/gist_id}","starred_url":"https://api.github.com/users/bharathkk/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/bharathkk/subscriptions","organizations_url":"https://api.github.com/users/bharathkk/orgs","repos_url":"https://api.github.com/users/bharathkk/repos","events_url":"https://api.github.com/users/bharathkk/events{/privacy}","received_events_url":"https://api.github.com/users/bharathkk/received_events","type":"User","site_admin":false},"body":"Getting the PR out to unblock sanil.\r\nPending tasks\r\n 1. Add documentation\r\n 2. Add more tests\r\n 3. Currently initialization of the stream happens using createStream on admin. We need to find a way to expose the same functionality to low level users.\r\n 4. To populate the initial stream, clients need to get a handle on producer and produce the messages. Alternatively, we can support serialization of source data and pass it as config to the system to initialize. We have a hook in place for this but not implemented it completely.\r\n  5. Clean up consumed data on the buffer based on the lowest offsets of the consumers.\r\n\r\nI will create JIRAs for all the pending tasks and fix them iteratively.\r\n\r\n@xinyuiscool ^^","created_at":"2018-03-28T19:21:27Z","updated_at":"2018-04-11T22:59:46Z","closed_at":null,"merged_at":null,"merge_commit_sha":"fde106f023ac148657abd21af46a9a0987d1e76c","assignee":null,"assignees":[],"requested_reviewers":[],"requested_teams":[],"labels":[],"milestone":null,"commits_url":"https://api.github.com/repos/apache/samza/pulls/459/commits","review_comments_url":"https://api.github.com/repos/apache/samza/pulls/459/comments","review_comment_url":"https://api.github.com/repos/apache/samza/pulls/comments{/number}","comments_url":"https://api.github.com/repos/apache/samza/issues/459/comments","statuses_url":"https://api.github.com/repos/apache/samza/statuses/3b4068914d906200070150fb6a72a342a1ac70ed","head":{"label":"bharathkk:single-node-testing","ref":"single-node-testing","sha":"3b4068914d906200070150fb6a72a342a1ac70ed","user":{"login":"bharathkk","id":1001798,"avatar_url":"https://avatars2.githubusercontent.com/u/1001798?v=4","gravatar_id":"","url":"https://api.github.com/users/bharathkk","html_url":"https://github.com/bharathkk","followers_url":"https://api.github.com/users/bharathkk/followers","following_url":"https://api.github.com/users/bharathkk/following{/other_user}","gists_url":"https://api.github.com/users/bharathkk/gists{/gist_id}","starred_url":"https://api.github.com/users/bharathkk/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/bharathkk/subscriptions","organizations_url":"https://api.github.com/users/bharathkk/orgs","repos_url":"https://api.github.com/users/bharathkk/repos","events_url":"https://api.github.com/users/bharathkk/events{/privacy}","received_events_url":"https://api.github.com/users/bharathkk/received_events","type":"User","site_admin":false},"repo":{"id":90763289,"name":"samza","full_name":"bharathkk/samza","owner":{"login":"bharathkk","id":1001798,"avatar_url":"https://avatars2.githubusercontent.com/u/1001798?v=4","gravatar_id":"","url":"https://api.github.com/users/bharathkk","html_url":"https://github.com/bharathkk","followers_url":"https://api.github.com/users/bharathkk/followers","following_url":"https://api.github.com/users/bharathkk/following{/other_user}","gists_url":"https://api.github.com/users/bharathkk/gists{/gist_id}","starred_url":"https://api.github.com/users/bharathkk/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/bharathkk/subscriptions","organizations_url":"https://api.github.com/users/bharathkk/orgs","repos_url":"https://api.github.com/users/bharathkk/repos","events_url":"https://api.github.com/users/bharathkk/events{/privacy}","received_events_url":"https://api.github.com/users/bharathkk/received_events","type":"User","site_admin":false},"private":false,"html_url":"https://github.com/bharathkk/samza","description":"Mirror of Apache Samza","fork":true,"url":"https://api.github.com/repos/bharathkk/samza","forks_url":"https://api.github.com/repos/bharathkk/samza/forks","keys_url":"https://api.github.com/repos/bharathkk/samza/keys{/key_id}","collaborators_url":"https://api.github.com/repos/bharathkk/samza/collaborators{/collaborator}","teams_url":"https://api.github.com/repos/bharathkk/samza/teams","hooks_url":"https://api.github.com/repos/bharathkk/samza/hooks","issue_events_url":"https://api.github.com/repos/bharathkk/samza/issues/events{/number}","events_url":"https://api.github.com/repos/bharathkk/samza/events","assignees_url":"https://api.github.com/repos/bharathkk/samza/assignees{/user}","branches_url":"https://api.github.com/repos/bharathkk/samza/branches{/branch}","tags_url":"https://api.github.com/repos/bharathkk/samza/tags","blobs_url":"https://api.github.com/repos/bharathkk/samza/git/blobs{/sha}","git_tags_url":"https://api.github.com/repos/bharathkk/samza/git/tags{/sha}","git_refs_url":"https://api.github.com/repos/bharathkk/samza/git/refs{/sha}","trees_url":"https://api.github.com/repos/bharathkk/samza/git/trees{/sha}","statuses_url":"https://api.github.com/repos/bharathkk/samza/statuses/{sha}","languages_url":"https://api.github.com/repos/bharathkk/samza/languages","stargazers_url":"https://api.github.com/repos/bharathkk/samza/stargazers","contributors_url":"https://api.github.com/repos/bharathkk/samza/contributors","subscribers_url":"https://api.github.com/repos/bharathkk/samza/subscribers","subscription_url":"https://api.github.com/repos/bharathkk/samza/subscription","commits_url":"https://api.github.com/repos/bharathkk/samza/commits{/sha}","git_commits_url":"https://api.github.com/repos/bharathkk/samza/git/commits{/sha}","comments_url":"https://api.github.com/repos/bharathkk/samza/comments{/number}","issue_comment_url":"https://api.github.com/repos/bharathkk/samza/issues/comments{/number}","contents_url":"https://api.github.com/repos/bharathkk/samza/contents/{+path}","compare_url":"https://api.github.com/repos/bharathkk/samza/compare/{base}...{head}","merges_url":"https://api.github.com/repos/bharathkk/samza/merges","archive_url":"https://api.github.com/repos/bharathkk/samza/{archive_format}{/ref}","downloads_url":"https://api.github.com/repos/bharathkk/samza/downloads","issues_url":"https://api.github.com/repos/bharathkk/samza/issues{/number}","pulls_url":"https://api.github.com/repos/bharathkk/samza/pulls{/number}","milestones_url":"https://api.github.com/repos/bharathkk/samza/milestones{/number}","notifications_url":"https://api.github.com/repos/bharathkk/samza/notifications{?since,all,participating}","labels_url":"https://api.github.com/repos/bharathkk/samza/labels{/name}","releases_url":"https://api.github.com/repos/bharathkk/samza/releases{/id}","deployments_url":"https://api.github.com/repos/bharathkk/samza/deployments","created_at":"2017-05-09T15:39:36Z","updated_at":"2018-03-28T18:53:03Z","pushed_at":"2018-03-29T00:34:06Z","git_url":"git://github.com/bharathkk/samza.git","ssh_url":"git@github.com:bharathkk/samza.git","clone_url":"https://github.com/bharathkk/samza.git","svn_url":"https://github.com/bharathkk/samza","homepage":null,"size":19290,"stargazers_count":0,"watchers_count":0,"language":"Java","has_issues":false,"has_projects":true,"has_downloads":true,"has_wiki":false,"has_pages":false,"forks_count":0,"mirror_url":null,"archived":false,"open_issues_count":0,"license":{"key":"apache-2.0","name":"Apache License 2.0","spdx_id":"Apache-2.0","url":"https://api.github.com/licenses/apache-2.0"},"forks":0,"open_issues":0,"watchers":0,"default_branch":"master"}},"base":{"label":"apache:master","ref":"master","sha":"5431350b7390704e395c947834b01a5f2e76d906","user":{"login":"apache","id":47359,"avatar_url":"https://avatars0.githubusercontent.com/u/47359?v=4","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":32199982,"name":"samza","full_name":"apache/samza","owner":{"login":"apache","id":47359,"avatar_url":"https://avatars0.githubusercontent.com/u/47359?v=4","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/samza","description":"Mirror of Apache Samza","fork":false,"url":"https://api.github.com/repos/apache/samza","forks_url":"https://api.github.com/repos/apache/samza/forks","keys_url":"https://api.github.com/repos/apache/samza/keys{/key_id}","collaborators_url":"https://api.github.com/repos/apache/samza/collaborators{/collaborator}","teams_url":"https://api.github.com/repos/apache/samza/teams","hooks_url":"https://api.github.com/repos/apache/samza/hooks","issue_events_url":"https://api.github.com/repos/apache/samza/issues/events{/number}","events_url":"https://api.github.com/repos/apache/samza/events","assignees_url":"https://api.github.com/repos/apache/samza/assignees{/user}","branches_url":"https://api.github.com/repos/apache/samza/branches{/branch}","tags_url":"https://api.github.com/repos/apache/samza/tags","blobs_url":"https://api.github.com/repos/apache/samza/git/blobs{/sha}","git_tags_url":"https://api.github.com/repos/apache/samza/git/tags{/sha}","git_refs_url":"https://api.github.com/repos/apache/samza/git/refs{/sha}","trees_url":"https://api.github.com/repos/apache/samza/git/trees{/sha}","statuses_url":"https://api.github.com/repos/apache/samza/statuses/{sha}","languages_url":"https://api.github.com/repos/apache/samza/languages","stargazers_url":"https://api.github.com/repos/apache/samza/stargazers","contributors_url":"https://api.github.com/repos/apache/samza/contributors","subscribers_url":"https://api.github.com/repos/apache/samza/subscribers","subscription_url":"https://api.github.com/repos/apache/samza/subscription","commits_url":"https://api.github.com/repos/apache/samza/commits{/sha}","git_commits_url":"https://api.github.com/repos/apache/samza/git/commits{/sha}","comments_url":"https://api.github.com/repos/apache/samza/comments{/number}","issue_comment_url":"https://api.github.com/repos/apache/samza/issues/comments{/number}","contents_url":"https://api.github.com/repos/apache/samza/contents/{+path}","compare_url":"https://api.github.com/repos/apache/samza/compare/{base}...{head}","merges_url":"https://api.github.com/repos/apache/samza/merges","archive_url":"https://api.github.com/repos/apache/samza/{archive_format}{/ref}","downloads_url":"https://api.github.com/repos/apache/samza/downloads","issues_url":"https://api.github.com/repos/apache/samza/issues{/number}","pulls_url":"https://api.github.com/repos/apache/samza/pulls{/number}","milestones_url":"https://api.github.com/repos/apache/samza/milestones{/number}","notifications_url":"https://api.github.com/repos/apache/samza/notifications{?since,all,participating}","labels_url":"https://api.github.com/repos/apache/samza/labels{/name}","releases_url":"https://api.github.com/repos/apache/samza/releases{/id}","deployments_url":"https://api.github.com/repos/apache/samza/deployments","created_at":"2015-03-14T07:00:05Z","updated_at":"2018-04-11T22:35:05Z","pushed_at":"2018-04-11T22:35:00Z","git_url":"git://github.com/apache/samza.git","ssh_url":"git@github.com:apache/samza.git","clone_url":"https://github.com/apache/samza.git","svn_url":"https://github.com/apache/samza","homepage":null,"size":19358,"stargazers_count":417,"watchers_count":417,"language":"Java","has_issues":false,"has_projects":true,"has_downloads":true,"has_wiki":false,"has_pages":false,"forks_count":183,"mirror_url":null,"archived":false,"open_issues_count":21,"license":{"key":"apache-2.0","name":"Apache License 2.0","spdx_id":"Apache-2.0","url":"https://api.github.com/licenses/apache-2.0"},"forks":183,"open_issues":21,"watchers":417,"default_branch":"master"}},"_links":{"self":{"href":"https://api.github.com/repos/apache/samza/pulls/459"},"html":{"href":"https://github.com/apache/samza/pull/459"},"issue":{"href":"https://api.github.com/repos/apache/samza/issues/459"},"comments":{"href":"https://api.github.com/repos/apache/samza/issues/459/comments"},"review_comments":{"href":"https://api.github.com/repos/apache/samza/pulls/459/comments"},"review_comment":{"href":"https://api.github.com/repos/apache/samza/pulls/comments{/number}"},"commits":{"href":"https://api.github.com/repos/apache/samza/pulls/459/commits"},"statuses":{"href":"https://api.github.com/repos/apache/samza/statuses/3b4068914d906200070150fb6a72a342a1ac70ed"}},"author_association":"NONE"}}
{"actor":{"display_login":"xinyuiscool"}}