{"action":"created","comment":{"url":"https://api.github.com/repos/prestodb/presto/pulls/comments/351534040","pull_request_review_id":323979166,"id":351534040,"node_id":"MDI0OlB1bGxSZXF1ZXN0UmV2aWV3Q29tbWVudDM1MTUzNDA0MA==","diff_hunk":"@@ -84,11 +85,54 @@\n \n         ImmutableList.Builder<Type> typeBuilder = ImmutableList.builder();\n \n-        for (DecoderColumnHandle handle : columnHandles) {\n+        for (DecoderColumnHandle handle : this.columnHandles) {\n             typeBuilder.add(handle.getType());\n         }\n \n         this.columnTypes = typeBuilder.build();\n+        KafkaConsumer consumer = consumerManager.createConsumer(Thread.currentThread().getId(), split.getLeader());\n+        TopicPartition tp = new TopicPartition(split.getTopicName(), split.getPartitionId());\n+        consumer.assign(ImmutableList.of(tp));\n+        setOffsetRange(consumer, split);\n+        consumer.close();\n+    }\n+\n+    private static void setOffsetRange(KafkaConsumer<Long, String> consumer, KafkaSplit split)\n+    {\n+        if (split.getStartTs() > split.getEndTs()) {\n+            throw new IllegalArgumentException(String.format(\"Invalid Kafka Offset start/end pair: %s - %s\", split.getStartTs(), split.getEndTs()));\n+        }\n+\n+        TopicPartition topicPartition = new TopicPartition(split.getTopicName(), split.getPartitionId());\n+        consumer.assign(ImmutableList.of(topicPartition));\n+\n+        long beginningOffset =\n+                (split.getStartTs() == 0) ?\n+                        consumer.beginningOffsets(ImmutableList.of(topicPartition)).values().iterator().next() :","path":"presto-kafka/src/main/java/com/facebook/presto/kafka/KafkaRecordSet.java","position":73,"original_position":73,"commit_id":"03d1d239a200b0771787a79818aa46e25de6ca5d","original_commit_id":"03d1d239a200b0771787a79818aa46e25de6ca5d","user":{"login":"zhenxiao","id":1274155,"node_id":"MDQ6VXNlcjEyNzQxNTU=","avatar_url":"https://avatars2.githubusercontent.com/u/1274155?v=4","gravatar_id":"","url":"https://api.github.com/users/zhenxiao","html_url":"https://github.com/zhenxiao","followers_url":"https://api.github.com/users/zhenxiao/followers","following_url":"https://api.github.com/users/zhenxiao/following{/other_user}","gists_url":"https://api.github.com/users/zhenxiao/gists{/gist_id}","starred_url":"https://api.github.com/users/zhenxiao/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/zhenxiao/subscriptions","organizations_url":"https://api.github.com/users/zhenxiao/orgs","repos_url":"https://api.github.com/users/zhenxiao/repos","events_url":"https://api.github.com/users/zhenxiao/events{/privacy}","received_events_url":"https://api.github.com/users/zhenxiao/received_events","type":"User","site_admin":false},"body":"could we use KafkaConsumer#position() to get the offset?","created_at":"2019-11-27T23:21:30Z","updated_at":"2019-11-27T23:53:10Z","html_url":"https://github.com/prestodb/presto/pull/13709#discussion_r351534040","pull_request_url":"https://api.github.com/repos/prestodb/presto/pulls/13709","author_association":"COLLABORATOR","_links":{"self":{"href":"https://api.github.com/repos/prestodb/presto/pulls/comments/351534040"},"html":{"href":"https://github.com/prestodb/presto/pull/13709#discussion_r351534040"},"pull_request":{"href":"https://api.github.com/repos/prestodb/presto/pulls/13709"}}},"pull_request":{"url":"https://api.github.com/repos/prestodb/presto/pulls/13709","id":341685172,"node_id":"MDExOlB1bGxSZXF1ZXN0MzQxNjg1MTcy","html_url":"https://github.com/prestodb/presto/pull/13709","diff_url":"https://github.com/prestodb/presto/pull/13709.diff","patch_url":"https://github.com/prestodb/presto/pull/13709.patch","issue_url":"https://api.github.com/repos/prestodb/presto/issues/13709","number":13709,"state":"open","locked":false,"title":"Update Kafka Connector","user":{"login":"dcfocus","id":29189795,"node_id":"MDQ6VXNlcjI5MTg5Nzk1","avatar_url":"https://avatars1.githubusercontent.com/u/29189795?v=4","gravatar_id":"","url":"https://api.github.com/users/dcfocus","html_url":"https://github.com/dcfocus","followers_url":"https://api.github.com/users/dcfocus/followers","following_url":"https://api.github.com/users/dcfocus/following{/other_user}","gists_url":"https://api.github.com/users/dcfocus/gists{/gist_id}","starred_url":"https://api.github.com/users/dcfocus/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/dcfocus/subscriptions","organizations_url":"https://api.github.com/users/dcfocus/orgs","repos_url":"https://api.github.com/users/dcfocus/repos","events_url":"https://api.github.com/users/dcfocus/events{/privacy}","received_events_url":"https://api.github.com/users/dcfocus/received_events","type":"User","site_admin":false},"body":"This PR improves Kafka connector implementation and performance.\r\n\r\n1. Replaced SimpleConsumer with KafkaConsumer for Kafka connector.\r\n2. Push collecting offsets for each partition from coordinator to workers to resolve the bottleneck.","created_at":"2019-11-15T23:18:02Z","updated_at":"2019-11-27T23:53:10Z","closed_at":null,"merged_at":null,"merge_commit_sha":"08919877a0ec4976968265984b3b03d3b1eca9a8","assignee":null,"assignees":[],"requested_reviewers":[],"requested_teams":[],"labels":[],"milestone":null,"commits_url":"https://api.github.com/repos/prestodb/presto/pulls/13709/commits","review_comments_url":"https://api.github.com/repos/prestodb/presto/pulls/13709/comments","review_comment_url":"https://api.github.com/repos/prestodb/presto/pulls/comments{/number}","comments_url":"https://api.github.com/repos/prestodb/presto/issues/13709/comments","statuses_url":"https://api.github.com/repos/prestodb/presto/statuses/03d1d239a200b0771787a79818aa46e25de6ca5d","head":{"label":"dcfocus:oss-multi-commits/upgrade-to-kafkaconsumer","ref":"oss-multi-commits/upgrade-to-kafkaconsumer","sha":"03d1d239a200b0771787a79818aa46e25de6ca5d","user":{"login":"dcfocus","id":29189795,"node_id":"MDQ6VXNlcjI5MTg5Nzk1","avatar_url":"https://avatars1.githubusercontent.com/u/29189795?v=4","gravatar_id":"","url":"https://api.github.com/users/dcfocus","html_url":"https://github.com/dcfocus","followers_url":"https://api.github.com/users/dcfocus/followers","following_url":"https://api.github.com/users/dcfocus/following{/other_user}","gists_url":"https://api.github.com/users/dcfocus/gists{/gist_id}","starred_url":"https://api.github.com/users/dcfocus/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/dcfocus/subscriptions","organizations_url":"https://api.github.com/users/dcfocus/orgs","repos_url":"https://api.github.com/users/dcfocus/repos","events_url":"https://api.github.com/users/dcfocus/events{/privacy}","received_events_url":"https://api.github.com/users/dcfocus/received_events","type":"User","site_admin":false},"repo":{"id":121807325,"node_id":"MDEwOlJlcG9zaXRvcnkxMjE4MDczMjU=","name":"presto","full_name":"dcfocus/presto","private":false,"owner":{"login":"dcfocus","id":29189795,"node_id":"MDQ6VXNlcjI5MTg5Nzk1","avatar_url":"https://avatars1.githubusercontent.com/u/29189795?v=4","gravatar_id":"","url":"https://api.github.com/users/dcfocus","html_url":"https://github.com/dcfocus","followers_url":"https://api.github.com/users/dcfocus/followers","following_url":"https://api.github.com/users/dcfocus/following{/other_user}","gists_url":"https://api.github.com/users/dcfocus/gists{/gist_id}","starred_url":"https://api.github.com/users/dcfocus/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/dcfocus/subscriptions","organizations_url":"https://api.github.com/users/dcfocus/orgs","repos_url":"https://api.github.com/users/dcfocus/repos","events_url":"https://api.github.com/users/dcfocus/events{/privacy}","received_events_url":"https://api.github.com/users/dcfocus/received_events","type":"User","site_admin":false},"html_url":"https://github.com/dcfocus/presto","description":"Distributed SQL query engine for running interactive analytic queries against big data sources.","fork":true,"url":"https://api.github.com/repos/dcfocus/presto","forks_url":"https://api.github.com/repos/dcfocus/presto/forks","keys_url":"https://api.github.com/repos/dcfocus/presto/keys{/key_id}","collaborators_url":"https://api.github.com/repos/dcfocus/presto/collaborators{/collaborator}","teams_url":"https://api.github.com/repos/dcfocus/presto/teams","hooks_url":"https://api.github.com/repos/dcfocus/presto/hooks","issue_events_url":"https://api.github.com/repos/dcfocus/presto/issues/events{/number}","events_url":"https://api.github.com/repos/dcfocus/presto/events","assignees_url":"https://api.github.com/repos/dcfocus/presto/assignees{/user}","branches_url":"https://api.github.com/repos/dcfocus/presto/branches{/branch}","tags_url":"https://api.github.com/repos/dcfocus/presto/tags","blobs_url":"https://api.github.com/repos/dcfocus/presto/git/blobs{/sha}","git_tags_url":"https://api.github.com/repos/dcfocus/presto/git/tags{/sha}","git_refs_url":"https://api.github.com/repos/dcfocus/presto/git/refs{/sha}","trees_url":"https://api.github.com/repos/dcfocus/presto/git/trees{/sha}","statuses_url":"https://api.github.com/repos/dcfocus/presto/statuses/{sha}","languages_url":"https://api.github.com/repos/dcfocus/presto/languages","stargazers_url":"https://api.github.com/repos/dcfocus/presto/stargazers","contributors_url":"https://api.github.com/repos/dcfocus/presto/contributors","subscribers_url":"https://api.github.com/repos/dcfocus/presto/subscribers","subscription_url":"https://api.github.com/repos/dcfocus/presto/subscription","commits_url":"https://api.github.com/repos/dcfocus/presto/commits{/sha}","git_commits_url":"https://api.github.com/repos/dcfocus/presto/git/commits{/sha}","comments_url":"https://api.github.com/repos/dcfocus/presto/comments{/number}","issue_comment_url":"https://api.github.com/repos/dcfocus/presto/issues/comments{/number}","contents_url":"https://api.github.com/repos/dcfocus/presto/contents/{+path}","compare_url":"https://api.github.com/repos/dcfocus/presto/compare/{base}...{head}","merges_url":"https://api.github.com/repos/dcfocus/presto/merges","archive_url":"https://api.github.com/repos/dcfocus/presto/{archive_format}{/ref}","downloads_url":"https://api.github.com/repos/dcfocus/presto/downloads","issues_url":"https://api.github.com/repos/dcfocus/presto/issues{/number}","pulls_url":"https://api.github.com/repos/dcfocus/presto/pulls{/number}","milestones_url":"https://api.github.com/repos/dcfocus/presto/milestones{/number}","notifications_url":"https://api.github.com/repos/dcfocus/presto/notifications{?since,all,participating}","labels_url":"https://api.github.com/repos/dcfocus/presto/labels{/name}","releases_url":"https://api.github.com/repos/dcfocus/presto/releases{/id}","deployments_url":"https://api.github.com/repos/dcfocus/presto/deployments","created_at":"2018-02-16T22:24:14Z","updated_at":"2019-04-26T23:17:37Z","pushed_at":"2019-11-21T05:40:03Z","git_url":"git://github.com/dcfocus/presto.git","ssh_url":"git@github.com:dcfocus/presto.git","clone_url":"https://github.com/dcfocus/presto.git","svn_url":"https://github.com/dcfocus/presto","homepage":"https://prestodb.io","size":120565,"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,"disabled":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","node_id":"MDc6TGljZW5zZTI="},"forks":0,"open_issues":0,"watchers":0,"default_branch":"twitter-master"}},"base":{"label":"prestodb:release-0.228","ref":"release-0.228","sha":"9da8c96ab81096cc79ae1bd7c37c8d21092992c2","user":{"login":"prestodb","id":6882181,"node_id":"MDEyOk9yZ2FuaXphdGlvbjY4ODIxODE=","avatar_url":"https://avatars0.githubusercontent.com/u/6882181?v=4","gravatar_id":"","url":"https://api.github.com/users/prestodb","html_url":"https://github.com/prestodb","followers_url":"https://api.github.com/users/prestodb/followers","following_url":"https://api.github.com/users/prestodb/following{/other_user}","gists_url":"https://api.github.com/users/prestodb/gists{/gist_id}","starred_url":"https://api.github.com/users/prestodb/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/prestodb/subscriptions","organizations_url":"https://api.github.com/users/prestodb/orgs","repos_url":"https://api.github.com/users/prestodb/repos","events_url":"https://api.github.com/users/prestodb/events{/privacy}","received_events_url":"https://api.github.com/users/prestodb/received_events","type":"Organization","site_admin":false},"repo":{"id":5349565,"node_id":"MDEwOlJlcG9zaXRvcnk1MzQ5NTY1","name":"presto","full_name":"prestodb/presto","private":false,"owner":{"login":"prestodb","id":6882181,"node_id":"MDEyOk9yZ2FuaXphdGlvbjY4ODIxODE=","avatar_url":"https://avatars0.githubusercontent.com/u/6882181?v=4","gravatar_id":"","url":"https://api.github.com/users/prestodb","html_url":"https://github.com/prestodb","followers_url":"https://api.github.com/users/prestodb/followers","following_url":"https://api.github.com/users/prestodb/following{/other_user}","gists_url":"https://api.github.com/users/prestodb/gists{/gist_id}","starred_url":"https://api.github.com/users/prestodb/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/prestodb/subscriptions","organizations_url":"https://api.github.com/users/prestodb/orgs","repos_url":"https://api.github.com/users/prestodb/repos","events_url":"https://api.github.com/users/prestodb/events{/privacy}","received_events_url":"https://api.github.com/users/prestodb/received_events","type":"Organization","site_admin":false},"html_url":"https://github.com/prestodb/presto","description":"The official home of the Presto distributed SQL query engine for big data","fork":false,"url":"https://api.github.com/repos/prestodb/presto","forks_url":"https://api.github.com/repos/prestodb/presto/forks","keys_url":"https://api.github.com/repos/prestodb/presto/keys{/key_id}","collaborators_url":"https://api.github.com/repos/prestodb/presto/collaborators{/collaborator}","teams_url":"https://api.github.com/repos/prestodb/presto/teams","hooks_url":"https://api.github.com/repos/prestodb/presto/hooks","issue_events_url":"https://api.github.com/repos/prestodb/presto/issues/events{/number}","events_url":"https://api.github.com/repos/prestodb/presto/events","assignees_url":"https://api.github.com/repos/prestodb/presto/assignees{/user}","branches_url":"https://api.github.com/repos/prestodb/presto/branches{/branch}","tags_url":"https://api.github.com/repos/prestodb/presto/tags","blobs_url":"https://api.github.com/repos/prestodb/presto/git/blobs{/sha}","git_tags_url":"https://api.github.com/repos/prestodb/presto/git/tags{/sha}","git_refs_url":"https://api.github.com/repos/prestodb/presto/git/refs{/sha}","trees_url":"https://api.github.com/repos/prestodb/presto/git/trees{/sha}","statuses_url":"https://api.github.com/repos/prestodb/presto/statuses/{sha}","languages_url":"https://api.github.com/repos/prestodb/presto/languages","stargazers_url":"https://api.github.com/repos/prestodb/presto/stargazers","contributors_url":"https://api.github.com/repos/prestodb/presto/contributors","subscribers_url":"https://api.github.com/repos/prestodb/presto/subscribers","subscription_url":"https://api.github.com/repos/prestodb/presto/subscription","commits_url":"https://api.github.com/repos/prestodb/presto/commits{/sha}","git_commits_url":"https://api.github.com/repos/prestodb/presto/git/commits{/sha}","comments_url":"https://api.github.com/repos/prestodb/presto/comments{/number}","issue_comment_url":"https://api.github.com/repos/prestodb/presto/issues/comments{/number}","contents_url":"https://api.github.com/repos/prestodb/presto/contents/{+path}","compare_url":"https://api.github.com/repos/prestodb/presto/compare/{base}...{head}","merges_url":"https://api.github.com/repos/prestodb/presto/merges","archive_url":"https://api.github.com/repos/prestodb/presto/{archive_format}{/ref}","downloads_url":"https://api.github.com/repos/prestodb/presto/downloads","issues_url":"https://api.github.com/repos/prestodb/presto/issues{/number}","pulls_url":"https://api.github.com/repos/prestodb/presto/pulls{/number}","milestones_url":"https://api.github.com/repos/prestodb/presto/milestones{/number}","notifications_url":"https://api.github.com/repos/prestodb/presto/notifications{?since,all,participating}","labels_url":"https://api.github.com/repos/prestodb/presto/labels{/name}","releases_url":"https://api.github.com/repos/prestodb/presto/releases{/id}","deployments_url":"https://api.github.com/repos/prestodb/presto/deployments","created_at":"2012-08-09T01:03:37Z","updated_at":"2019-11-27T23:35:03Z","pushed_at":"2019-11-27T23:51:53Z","git_url":"git://github.com/prestodb/presto.git","ssh_url":"git@github.com:prestodb/presto.git","clone_url":"https://github.com/prestodb/presto.git","svn_url":"https://github.com/prestodb/presto","homepage":"http://prestodb.github.io","size":123322,"stargazers_count":9914,"watchers_count":9914,"language":"Java","has_issues":true,"has_projects":true,"has_downloads":true,"has_wiki":true,"has_pages":false,"forks_count":3392,"mirror_url":null,"archived":false,"disabled":false,"open_issues_count":984,"license":{"key":"apache-2.0","name":"Apache License 2.0","spdx_id":"Apache-2.0","url":"https://api.github.com/licenses/apache-2.0","node_id":"MDc6TGljZW5zZTI="},"forks":3392,"open_issues":984,"watchers":9914,"default_branch":"master"}},"_links":{"self":{"href":"https://api.github.com/repos/prestodb/presto/pulls/13709"},"html":{"href":"https://github.com/prestodb/presto/pull/13709"},"issue":{"href":"https://api.github.com/repos/prestodb/presto/issues/13709"},"comments":{"href":"https://api.github.com/repos/prestodb/presto/issues/13709/comments"},"review_comments":{"href":"https://api.github.com/repos/prestodb/presto/pulls/13709/comments"},"review_comment":{"href":"https://api.github.com/repos/prestodb/presto/pulls/comments{/number}"},"commits":{"href":"https://api.github.com/repos/prestodb/presto/pulls/13709/commits"},"statuses":{"href":"https://api.github.com/repos/prestodb/presto/statuses/03d1d239a200b0771787a79818aa46e25de6ca5d"}},"author_association":"NONE"}}
{"actor":{"display_login":"zhenxiao"}}