{"action":"created","comment":{"url":"https://api.github.com/repos/apache/kafka/pulls/comments/39090892","id":39090892,"diff_hunk":"@@ -494,4 +497,43 @@ public static String readFileAsString(String path, Charset charset) throws IOExc\n     public static String readFileAsString(String path) throws IOException {\n         return Utils.readFileAsString(path, Charset.defaultCharset());\n     }\n+\n+    /**\n+     * Creates a set\n+     * @param elems the elements\n+     * @param <T> the type of element\n+     * @return Set\n+     */\n+    public static <T> HashSet<T> mkSet(T... elems) {\n+        return new HashSet<>(Arrays.asList(elems));\n+    }\n+    \n+    /**\n+     * Recursively delete the given file/directory and any subfiles (if any exist)\n+     *\n+     * @param file The root file at which to begin deleting\n+     */\n+    public static void rm(File file) {\n+        if (file == null) {\n+            return;\n+        } else if (file.isDirectory()) {\n+            File[] files = file.listFiles();\n+            if (files != null) {\n+                for (File f : files)\n+                    rm(f);\n+            }\n+            file.delete();\n+        } else {\n+            file.delete();\n+        }\n+    }\n+\n+    /**\n+     * Returns an empty collection if this list is empty\n+     * @param other\n+     * @return\n+     */\n+    public static <T> List<T> safe(List<T> other) {\n+        return other == null ? Collections.EMPTY_LIST : other;","path":"clients/src/main/java/org/apache/kafka/common/utils/Utils.java","position":52,"original_position":52,"commit_id":"279676d958ccb1723da4396533d5e3916c15343c","original_commit_id":"9287c20f88284cde4a182476f86af64883364e43","user":{"login":"guozhangwang","id":10729597,"avatar_url":"https://avatars.githubusercontent.com/u/10729597?v=3","gravatar_id":"","url":"https://api.github.com/users/guozhangwang","html_url":"https://github.com/guozhangwang","followers_url":"https://api.github.com/users/guozhangwang/followers","following_url":"https://api.github.com/users/guozhangwang/following{/other_user}","gists_url":"https://api.github.com/users/guozhangwang/gists{/gist_id}","starred_url":"https://api.github.com/users/guozhangwang/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/guozhangwang/subscriptions","organizations_url":"https://api.github.com/users/guozhangwang/orgs","repos_url":"https://api.github.com/users/guozhangwang/repos","events_url":"https://api.github.com/users/guozhangwang/events{/privacy}","received_events_url":"https://api.github.com/users/guozhangwang/received_events","type":"User","site_admin":false},"body":"Good point, changed to ``Collections.emptyList()``.\r\n\r\nThe semantics is only to return an empty list if ``other`` is null, if it is empty then we will still return itself.","created_at":"2015-09-09T20:20:34Z","updated_at":"2015-09-09T20:20:34Z","html_url":"https://github.com/apache/kafka/pull/130#discussion_r39090892","pull_request_url":"https://api.github.com/repos/apache/kafka/pulls/130","_links":{"self":{"href":"https://api.github.com/repos/apache/kafka/pulls/comments/39090892"},"html":{"href":"https://github.com/apache/kafka/pull/130#discussion_r39090892"},"pull_request":{"href":"https://api.github.com/repos/apache/kafka/pulls/130"}}},"pull_request":{"url":"https://api.github.com/repos/apache/kafka/pulls/130","id":42085063,"html_url":"https://github.com/apache/kafka/pull/130","diff_url":"https://github.com/apache/kafka/pull/130.diff","patch_url":"https://github.com/apache/kafka/pull/130.patch","issue_url":"https://api.github.com/repos/apache/kafka/issues/130","number":130,"state":"open","locked":false,"title":"KIP-28: First patch","user":{"login":"guozhangwang","id":10729597,"avatar_url":"https://avatars.githubusercontent.com/u/10729597?v=3","gravatar_id":"","url":"https://api.github.com/users/guozhangwang","html_url":"https://github.com/guozhangwang","followers_url":"https://api.github.com/users/guozhangwang/followers","following_url":"https://api.github.com/users/guozhangwang/following{/other_user}","gists_url":"https://api.github.com/users/guozhangwang/gists{/gist_id}","starred_url":"https://api.github.com/users/guozhangwang/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/guozhangwang/subscriptions","organizations_url":"https://api.github.com/users/guozhangwang/orgs","repos_url":"https://api.github.com/users/guozhangwang/repos","events_url":"https://api.github.com/users/guozhangwang/events{/privacy}","received_events_url":"https://api.github.com/users/guozhangwang/received_events","type":"User","site_admin":false},"body":"Some open questions collected so far on the first patch. Thanks @gwenshap @jkreps @junrao .\r\n\r\n* Topology API: requiring users to instantiate their own Topology class with the overridden build() function is a little awkward. Instead it would be great to let users explicitly build the topology in Main and pass it in as a class:\r\n\r\n```\r\n    Topology myTopology = new TopologyBuilder(defaultDeser)\r\n                                                 .addProcessor(\"my-processor\", MyProcessor.class, new Source(\"my-source\"))\r\n                                                 .addProcessor(\"my-other-processor\", MyOtherProcessor.class, \"my-processor\");\r\n    KafkaStreaming streaming = new KafkaStreaming(config, myTopology);\r\n   streaming.run();\r\n```\r\n\r\nSo the implementation of KStream.filter look instead like this:\r\n```\r\n    public KStream<K, V> filter(Predicate<K, V> predicate) {\r\n        KStreamFilter<K, V> filter = new KStreamFilter<>();\r\n        topology.addProcessor(KStreamFilter.class, new Configs(\"predicate\", predicate));\r\n        return this;\r\n    }\r\n```\r\nThe advantage is that the user code can now get rid of the whole Topology class with the builder. I think the order of execution for that API is quite unintuitive.\r\n\r\n* We can probably move the forward() function from Processor to ProcessorContext, and split ProcessorContext into two classes, one with all the function calls as commit / send / schedule / forward, and another with the metadata function calls as topic / partition / offset / timestamp.\r\n\r\n* Can we hide the Chooser interface from users? In other words, if users can specify the \"time\" on each fetched messages from Kafka, would a hard-coded MinTimestampMessageChooser be sufficient so that we can move TimestampTracker / RecordQueue / Chooser / RecordCollector / etc all to the internal folders?\r\n\r\n* Shall we split the o.a.k.clients into two folders, with o.a.k.clients.processor in stream? Or should we just remove o.a.k.clients.processor and make everything under o.a.k.stream? In addition, currently there is a cyclic dependency between that two, would better to break it in the end state.\r\n\r\n* Consider moving the external dependencies such as RocksDB into a separate jar? For example we can just include a kafka-stream-rocksdb.jar which includes the RocksDBKeyValueStore only, and later on when we deprecate / remove such implementations we can simply remove the jar itself.\r\n\r\n* The way the KeyValueStore is created seems a bit weird. Since this is part of the internal state managed by KafkaProcessorContext, it seems there should be an api to create the KeyValueStore from KafkaProcessorContext, instead of passing context to the constructor of KeyValueStore.\r\n\r\n* Merge ProcessorConfigs with ProcessorProperties.\r\n\r\n* We can potentially remove the processor argument in ProcessorContext.schedule().","created_at":"2015-08-11T01:45:03Z","updated_at":"2015-09-09T20:20:34Z","closed_at":null,"merged_at":null,"merge_commit_sha":"0f15c5896f438b81db91f3dab220c23423cf6975","assignee":null,"milestone":null,"commits_url":"https://api.github.com/repos/apache/kafka/pulls/130/commits","review_comments_url":"https://api.github.com/repos/apache/kafka/pulls/130/comments","review_comment_url":"https://api.github.com/repos/apache/kafka/pulls/comments{/number}","comments_url":"https://api.github.com/repos/apache/kafka/issues/130/comments","statuses_url":"https://api.github.com/repos/apache/kafka/statuses/279676d958ccb1723da4396533d5e3916c15343c","head":{"label":"confluentinc:streaming","ref":"streaming","sha":"279676d958ccb1723da4396533d5e3916c15343c","user":{"login":"confluentinc","id":9439498,"avatar_url":"https://avatars.githubusercontent.com/u/9439498?v=3","gravatar_id":"","url":"https://api.github.com/users/confluentinc","html_url":"https://github.com/confluentinc","followers_url":"https://api.github.com/users/confluentinc/followers","following_url":"https://api.github.com/users/confluentinc/following{/other_user}","gists_url":"https://api.github.com/users/confluentinc/gists{/gist_id}","starred_url":"https://api.github.com/users/confluentinc/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/confluentinc/subscriptions","organizations_url":"https://api.github.com/users/confluentinc/orgs","repos_url":"https://api.github.com/users/confluentinc/repos","events_url":"https://api.github.com/users/confluentinc/events{/privacy}","received_events_url":"https://api.github.com/users/confluentinc/received_events","type":"Organization","site_admin":false},"repo":{"id":37555321,"name":"kafka","full_name":"confluentinc/kafka","owner":{"login":"confluentinc","id":9439498,"avatar_url":"https://avatars.githubusercontent.com/u/9439498?v=3","gravatar_id":"","url":"https://api.github.com/users/confluentinc","html_url":"https://github.com/confluentinc","followers_url":"https://api.github.com/users/confluentinc/followers","following_url":"https://api.github.com/users/confluentinc/following{/other_user}","gists_url":"https://api.github.com/users/confluentinc/gists{/gist_id}","starred_url":"https://api.github.com/users/confluentinc/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/confluentinc/subscriptions","organizations_url":"https://api.github.com/users/confluentinc/orgs","repos_url":"https://api.github.com/users/confluentinc/repos","events_url":"https://api.github.com/users/confluentinc/events{/privacy}","received_events_url":"https://api.github.com/users/confluentinc/received_events","type":"Organization","site_admin":false},"private":false,"html_url":"https://github.com/confluentinc/kafka","description":"Mirror of Apache Kafka","fork":true,"url":"https://api.github.com/repos/confluentinc/kafka","forks_url":"https://api.github.com/repos/confluentinc/kafka/forks","keys_url":"https://api.github.com/repos/confluentinc/kafka/keys{/key_id}","collaborators_url":"https://api.github.com/repos/confluentinc/kafka/collaborators{/collaborator}","teams_url":"https://api.github.com/repos/confluentinc/kafka/teams","hooks_url":"https://api.github.com/repos/confluentinc/kafka/hooks","issue_events_url":"https://api.github.com/repos/confluentinc/kafka/issues/events{/number}","events_url":"https://api.github.com/repos/confluentinc/kafka/events","assignees_url":"https://api.github.com/repos/confluentinc/kafka/assignees{/user}","branches_url":"https://api.github.com/repos/confluentinc/kafka/branches{/branch}","tags_url":"https://api.github.com/repos/confluentinc/kafka/tags","blobs_url":"https://api.github.com/repos/confluentinc/kafka/git/blobs{/sha}","git_tags_url":"https://api.github.com/repos/confluentinc/kafka/git/tags{/sha}","git_refs_url":"https://api.github.com/repos/confluentinc/kafka/git/refs{/sha}","trees_url":"https://api.github.com/repos/confluentinc/kafka/git/trees{/sha}","statuses_url":"https://api.github.com/repos/confluentinc/kafka/statuses/{sha}","languages_url":"https://api.github.com/repos/confluentinc/kafka/languages","stargazers_url":"https://api.github.com/repos/confluentinc/kafka/stargazers","contributors_url":"https://api.github.com/repos/confluentinc/kafka/contributors","subscribers_url":"https://api.github.com/repos/confluentinc/kafka/subscribers","subscription_url":"https://api.github.com/repos/confluentinc/kafka/subscription","commits_url":"https://api.github.com/repos/confluentinc/kafka/commits{/sha}","git_commits_url":"https://api.github.com/repos/confluentinc/kafka/git/commits{/sha}","comments_url":"https://api.github.com/repos/confluentinc/kafka/comments{/number}","issue_comment_url":"https://api.github.com/repos/confluentinc/kafka/issues/comments{/number}","contents_url":"https://api.github.com/repos/confluentinc/kafka/contents/{+path}","compare_url":"https://api.github.com/repos/confluentinc/kafka/compare/{base}...{head}","merges_url":"https://api.github.com/repos/confluentinc/kafka/merges","archive_url":"https://api.github.com/repos/confluentinc/kafka/{archive_format}{/ref}","downloads_url":"https://api.github.com/repos/confluentinc/kafka/downloads","issues_url":"https://api.github.com/repos/confluentinc/kafka/issues{/number}","pulls_url":"https://api.github.com/repos/confluentinc/kafka/pulls{/number}","milestones_url":"https://api.github.com/repos/confluentinc/kafka/milestones{/number}","notifications_url":"https://api.github.com/repos/confluentinc/kafka/notifications{?since,all,participating}","labels_url":"https://api.github.com/repos/confluentinc/kafka/labels{/name}","releases_url":"https://api.github.com/repos/confluentinc/kafka/releases{/id}","created_at":"2015-06-16T20:48:28Z","updated_at":"2015-09-08T03:39:19Z","pushed_at":"2015-09-09T20:01:37Z","git_url":"git://github.com/confluentinc/kafka.git","ssh_url":"git@github.com:confluentinc/kafka.git","clone_url":"https://github.com/confluentinc/kafka.git","svn_url":"https://github.com/confluentinc/kafka","homepage":null,"size":23014,"stargazers_count":5,"watchers_count":5,"language":"Scala","has_issues":true,"has_downloads":true,"has_wiki":false,"has_pages":false,"forks_count":2,"mirror_url":null,"open_issues_count":1,"forks":2,"open_issues":1,"watchers":5,"default_branch":"trunk"}},"base":{"label":"apache:trunk","ref":"trunk","sha":"1ed88f0eba860bce4be0269cdd4d38c2b57db7f7","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}","created_at":"2011-08-15T18:06:16Z","updated_at":"2015-09-09T15:58:14Z","pushed_at":"2015-09-09T20:01:38Z","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":33192,"stargazers_count":1890,"watchers_count":1890,"language":"Scala","has_issues":false,"has_downloads":true,"has_wiki":false,"has_pages":false,"forks_count":1136,"mirror_url":"git://git.apache.org/kafka.git","open_issues_count":78,"forks":1136,"open_issues":78,"watchers":1890,"default_branch":"trunk"}},"_links":{"self":{"href":"https://api.github.com/repos/apache/kafka/pulls/130"},"html":{"href":"https://github.com/apache/kafka/pull/130"},"issue":{"href":"https://api.github.com/repos/apache/kafka/issues/130"},"comments":{"href":"https://api.github.com/repos/apache/kafka/issues/130/comments"},"review_comments":{"href":"https://api.github.com/repos/apache/kafka/pulls/130/comments"},"review_comment":{"href":"https://api.github.com/repos/apache/kafka/pulls/comments{/number}"},"commits":{"href":"https://api.github.com/repos/apache/kafka/pulls/130/commits"},"statuses":{"href":"https://api.github.com/repos/apache/kafka/statuses/279676d958ccb1723da4396533d5e3916c15343c"}}}}