{"action":"created","comment":{"url":"https://api.github.com/repos/apache/flink/pulls/comments/89365136","pull_request_review_id":9911327,"id":89365136,"diff_hunk":"@@ -139,24 +143,65 @@ public void runFetchLoop() throws Exception {\n \n \t\tPeriodicOffsetCommitter periodicCommitter = null;\n \t\ttry {\n-\t\t\t// read offsets from ZooKeeper for partitions that did not restore offsets\n-\t\t\t{\n-\t\t\t\tList<KafkaTopicPartition> partitionsWithNoOffset = new ArrayList<>();\n-\t\t\t\tfor (KafkaTopicPartitionState<TopicAndPartition> partition : subscribedPartitions()) {\n-\t\t\t\t\tif (!partition.isOffsetDefined()) {\n-\t\t\t\t\t\tpartitionsWithNoOffset.add(partition.getKafkaTopicPartition());\n-\t\t\t\t\t}\n+\t\t\tList<KafkaTopicPartition> partitionsWithNoOffset = new ArrayList<>();\n+\t\t\tfor (KafkaTopicPartitionState<TopicAndPartition> partition : subscribedPartitions()) {\n+\t\t\t\tif (!partition.isOffsetDefined()) {\n+\t\t\t\t\tpartitionsWithNoOffset.add(partition.getKafkaTopicPartition());\n \t\t\t\t}\n+\t\t\t}\n+\n+\t\t\tif (partitionsWithNoOffset.size() == subscribedPartitions().length) {\n+\t\t\t\t// if all partitions have no initial offsets, that means we're starting fresh without any restored state\n+\t\t\t\tswitch (startupMode) {\n+\t\t\t\t\tcase EARLIEST:\n+\t\t\t\t\t\tLOG.info(\"Setting starting point as earliest offset for partitions {}\", partitionsWithNoOffset);\n+\n+\t\t\t\t\t\tfor (KafkaTopicPartitionState<TopicAndPartition> partition : subscribedPartitions()) {\n+\t\t\t\t\t\t\tpartition.setOffset(OffsetRequest.EarliestTime());\n+\t\t\t\t\t\t}\n+\t\t\t\t\t\tbreak;\n+\t\t\t\t\tcase LATEST:\n+\t\t\t\t\t\tLOG.info(\"Setting starting point as latest offset for partitions {}\", partitionsWithNoOffset);\n+\n+\t\t\t\t\t\tfor (KafkaTopicPartitionState<TopicAndPartition> partition : subscribedPartitions()) {\n+\t\t\t\t\t\t\tpartition.setOffset(OffsetRequest.LatestTime());\n+\t\t\t\t\t\t}\n+\t\t\t\t\t\tbreak;\n+\t\t\t\t\tdefault:\n+\t\t\t\t\tcase GROUP_OFFSETS:\n+\t\t\t\t\t\tLOG.info(\"Using group offsets in Zookeeper of group.id {} as starting point for partitions {}\",\n+\t\t\t\t\t\t\tkafkaConfig.getProperty(\"group.id\"), partitionsWithNoOffset);\n+\n+\t\t\t\t\t\tMap<KafkaTopicPartition, Long> zkOffsets = zookeeperOffsetHandler.getCommittedOffsets(partitionsWithNoOffset);\n+\t\t\t\t\t\tfor (KafkaTopicPartitionState<TopicAndPartition> partition : subscribedPartitions()) {\n+\t\t\t\t\t\t\tLong offset = zkOffsets.get(partition.getKafkaTopicPartition());\n+\t\t\t\t\t\t\tif (offset != null) {\n+\t\t\t\t\t\t\t\t// the committed offset in ZK represents the next record to process,\n+\t\t\t\t\t\t\t\t// so we subtract it by 1 to correctly represent internal state\n+\t\t\t\t\t\t\t\tpartition.setOffset(offset - 1);\n+\t\t\t\t\t\t\t} else {\n+\t\t\t\t\t\t\t\t// if we can't find an offset for a partition in ZK when using GROUP_OFFSETS,\n+\t\t\t\t\t\t\t\t// we default to \"auto.offset.reset\" like the Kafka high-level consumer\n+\t\t\t\t\t\t\t\tLOG.warn(\"No group offset can be found for partition {} in Zookeeper;\" +\n+\t\t\t\t\t\t\t\t\t\" resetting starting offset to 'auto.offset.reset'\", partition);\n+\n+\t\t\t\t\t\t\t\tpartition.setOffset(invalidOffsetBehavior);\n+\t\t\t\t\t\t\t}\n+\t\t\t\t\t\t}\n+\t\t\t\t}\n+\t\t\t} else if (partitionsWithNoOffset.size() > 0 && partitionsWithNoOffset.size() < subscribedPartitions().length) {","path":"flink-streaming-connectors/flink-connector-kafka-0.8/src/main/java/org/apache/flink/streaming/connectors/kafka/internals/Kafka08Fetcher.java","position":95,"original_position":95,"commit_id":"d8f5f976ef2e2e1d994a45468d7e9ef3b8bf0015","original_commit_id":"d8f5f976ef2e2e1d994a45468d7e9ef3b8bf0015","user":{"login":"tzulitai","id":5284370,"avatar_url":"https://avatars.githubusercontent.com/u/5284370?v=3","gravatar_id":"","url":"https://api.github.com/users/tzulitai","html_url":"https://github.com/tzulitai","followers_url":"https://api.github.com/users/tzulitai/followers","following_url":"https://api.github.com/users/tzulitai/following{/other_user}","gists_url":"https://api.github.com/users/tzulitai/gists{/gist_id}","starred_url":"https://api.github.com/users/tzulitai/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/tzulitai/subscriptions","organizations_url":"https://api.github.com/users/tzulitai/orgs","repos_url":"https://api.github.com/users/tzulitai/repos","events_url":"https://api.github.com/users/tzulitai/events{/privacy}","received_events_url":"https://api.github.com/users/tzulitai/received_events","type":"User","site_admin":false},"body":"I was adding this as a preparation for the kafka partition discovery task.\r\nBut it'd probably make sense to remove it for this PR to avoid confusion.","created_at":"2016-11-23T17:36:42Z","updated_at":"2016-11-23T17:36:43Z","html_url":"https://github.com/apache/flink/pull/2509#discussion_r89365136","pull_request_url":"https://api.github.com/repos/apache/flink/pulls/2509","_links":{"self":{"href":"https://api.github.com/repos/apache/flink/pulls/comments/89365136"},"html":{"href":"https://github.com/apache/flink/pull/2509#discussion_r89365136"},"pull_request":{"href":"https://api.github.com/repos/apache/flink/pulls/2509"}}},"pull_request":{"url":"https://api.github.com/repos/apache/flink/pulls/2509","id":85676081,"html_url":"https://github.com/apache/flink/pull/2509","diff_url":"https://github.com/apache/flink/pull/2509.diff","patch_url":"https://github.com/apache/flink/pull/2509.patch","issue_url":"https://api.github.com/repos/apache/flink/issues/2509","number":2509,"state":"open","locked":false,"title":"[FLINK-4280][kafka-connector] Explicit start position configuration for Kafka Consumer","user":{"login":"tzulitai","id":5284370,"avatar_url":"https://avatars.githubusercontent.com/u/5284370?v=3","gravatar_id":"","url":"https://api.github.com/users/tzulitai","html_url":"https://github.com/tzulitai","followers_url":"https://api.github.com/users/tzulitai/followers","following_url":"https://api.github.com/users/tzulitai/following{/other_user}","gists_url":"https://api.github.com/users/tzulitai/gists{/gist_id}","starred_url":"https://api.github.com/users/tzulitai/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/tzulitai/subscriptions","organizations_url":"https://api.github.com/users/tzulitai/orgs","repos_url":"https://api.github.com/users/tzulitai/repos","events_url":"https://api.github.com/users/tzulitai/events{/privacy}","received_events_url":"https://api.github.com/users/tzulitai/received_events","type":"User","site_admin":false},"body":"This is the first PR for a series of new explicit setter configurations for the Kafka Consumer (see also [FLINK-3398](https://issues.apache.org/jira/browse/FLINK-3398) for details on opt-out from offset committing).\n\nIt adds the following new explicit setter methods to configure the starting position for the Kafka Consumer connector:\n\n```\nFlinkKafkaConsumer08 kafka = new FlinkKafkaConsumer08(...) // or 09\nkafka.setStartFromEarliest(); // start from earliest without respecting any committed offsets\nkafka.setStartFromLatest(); // start from latest without respecting any committed offsets\nkafka.setStartFromGroupOffsets(); // respects committed offsets in ZK / Kafka as starting points\n```\n\nThe default is to start from group offsets, so we won't be breaking existing user code.\n\nOne thing to note is that this PR also includes some refactoring to consolidate all start offset assigning logic for partitions within the fetcher. For example, in 0.8 version, with this change the `SimpleConsumerThread` no longer deals with deciding where a partition needs to start from; all partitions should already be assigned starting offsets by the fetcher, and it simply needs to start consuming the partition.This is a pre-preparation for transparent partition discovery for the Kafka consumers in [FLINK-4022](https://issues.apache.org/jira/browse/FLINK-4022).\n\nI suggest to review this PR after #2369 to reduce effort in getting the 0.10 Kafka consumer in first. Tests for the new function will be added in follow-up commits.\n","created_at":"2016-09-17T14:35:58Z","updated_at":"2016-11-23T17:36:43Z","closed_at":null,"merged_at":null,"merge_commit_sha":"f76cf0d1284c6ae4d5debfe3a8a79443b2bfd6e8","assignee":null,"assignees":[],"milestone":null,"commits_url":"https://api.github.com/repos/apache/flink/pulls/2509/commits","review_comments_url":"https://api.github.com/repos/apache/flink/pulls/2509/comments","review_comment_url":"https://api.github.com/repos/apache/flink/pulls/comments{/number}","comments_url":"https://api.github.com/repos/apache/flink/issues/2509/comments","statuses_url":"https://api.github.com/repos/apache/flink/statuses/d8f5f976ef2e2e1d994a45468d7e9ef3b8bf0015","head":{"label":"tzulitai:FLINK-4280","ref":"FLINK-4280","sha":"d8f5f976ef2e2e1d994a45468d7e9ef3b8bf0015","user":{"login":"tzulitai","id":5284370,"avatar_url":"https://avatars.githubusercontent.com/u/5284370?v=3","gravatar_id":"","url":"https://api.github.com/users/tzulitai","html_url":"https://github.com/tzulitai","followers_url":"https://api.github.com/users/tzulitai/followers","following_url":"https://api.github.com/users/tzulitai/following{/other_user}","gists_url":"https://api.github.com/users/tzulitai/gists{/gist_id}","starred_url":"https://api.github.com/users/tzulitai/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/tzulitai/subscriptions","organizations_url":"https://api.github.com/users/tzulitai/orgs","repos_url":"https://api.github.com/users/tzulitai/repos","events_url":"https://api.github.com/users/tzulitai/events{/privacy}","received_events_url":"https://api.github.com/users/tzulitai/received_events","type":"User","site_admin":false},"repo":{"id":55886834,"name":"flink","full_name":"tzulitai/flink","owner":{"login":"tzulitai","id":5284370,"avatar_url":"https://avatars.githubusercontent.com/u/5284370?v=3","gravatar_id":"","url":"https://api.github.com/users/tzulitai","html_url":"https://github.com/tzulitai","followers_url":"https://api.github.com/users/tzulitai/followers","following_url":"https://api.github.com/users/tzulitai/following{/other_user}","gists_url":"https://api.github.com/users/tzulitai/gists{/gist_id}","starred_url":"https://api.github.com/users/tzulitai/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/tzulitai/subscriptions","organizations_url":"https://api.github.com/users/tzulitai/orgs","repos_url":"https://api.github.com/users/tzulitai/repos","events_url":"https://api.github.com/users/tzulitai/events{/privacy}","received_events_url":"https://api.github.com/users/tzulitai/received_events","type":"User","site_admin":false},"private":false,"html_url":"https://github.com/tzulitai/flink","description":"Mirror of Apache Flink","fork":true,"url":"https://api.github.com/repos/tzulitai/flink","forks_url":"https://api.github.com/repos/tzulitai/flink/forks","keys_url":"https://api.github.com/repos/tzulitai/flink/keys{/key_id}","collaborators_url":"https://api.github.com/repos/tzulitai/flink/collaborators{/collaborator}","teams_url":"https://api.github.com/repos/tzulitai/flink/teams","hooks_url":"https://api.github.com/repos/tzulitai/flink/hooks","issue_events_url":"https://api.github.com/repos/tzulitai/flink/issues/events{/number}","events_url":"https://api.github.com/repos/tzulitai/flink/events","assignees_url":"https://api.github.com/repos/tzulitai/flink/assignees{/user}","branches_url":"https://api.github.com/repos/tzulitai/flink/branches{/branch}","tags_url":"https://api.github.com/repos/tzulitai/flink/tags","blobs_url":"https://api.github.com/repos/tzulitai/flink/git/blobs{/sha}","git_tags_url":"https://api.github.com/repos/tzulitai/flink/git/tags{/sha}","git_refs_url":"https://api.github.com/repos/tzulitai/flink/git/refs{/sha}","trees_url":"https://api.github.com/repos/tzulitai/flink/git/trees{/sha}","statuses_url":"https://api.github.com/repos/tzulitai/flink/statuses/{sha}","languages_url":"https://api.github.com/repos/tzulitai/flink/languages","stargazers_url":"https://api.github.com/repos/tzulitai/flink/stargazers","contributors_url":"https://api.github.com/repos/tzulitai/flink/contributors","subscribers_url":"https://api.github.com/repos/tzulitai/flink/subscribers","subscription_url":"https://api.github.com/repos/tzulitai/flink/subscription","commits_url":"https://api.github.com/repos/tzulitai/flink/commits{/sha}","git_commits_url":"https://api.github.com/repos/tzulitai/flink/git/commits{/sha}","comments_url":"https://api.github.com/repos/tzulitai/flink/comments{/number}","issue_comment_url":"https://api.github.com/repos/tzulitai/flink/issues/comments{/number}","contents_url":"https://api.github.com/repos/tzulitai/flink/contents/{+path}","compare_url":"https://api.github.com/repos/tzulitai/flink/compare/{base}...{head}","merges_url":"https://api.github.com/repos/tzulitai/flink/merges","archive_url":"https://api.github.com/repos/tzulitai/flink/{archive_format}{/ref}","downloads_url":"https://api.github.com/repos/tzulitai/flink/downloads","issues_url":"https://api.github.com/repos/tzulitai/flink/issues{/number}","pulls_url":"https://api.github.com/repos/tzulitai/flink/pulls{/number}","milestones_url":"https://api.github.com/repos/tzulitai/flink/milestones{/number}","notifications_url":"https://api.github.com/repos/tzulitai/flink/notifications{?since,all,participating}","labels_url":"https://api.github.com/repos/tzulitai/flink/labels{/name}","releases_url":"https://api.github.com/repos/tzulitai/flink/releases{/id}","deployments_url":"https://api.github.com/repos/tzulitai/flink/deployments","created_at":"2016-04-10T07:41:34Z","updated_at":"2016-04-10T07:41:51Z","pushed_at":"2016-11-22T11:37:40Z","git_url":"git://github.com/tzulitai/flink.git","ssh_url":"git@github.com:tzulitai/flink.git","clone_url":"https://github.com/tzulitai/flink.git","svn_url":"https://github.com/tzulitai/flink","homepage":null,"size":102234,"stargazers_count":0,"watchers_count":0,"language":"Java","has_issues":false,"has_downloads":true,"has_wiki":false,"has_pages":false,"forks_count":1,"mirror_url":null,"open_issues_count":0,"forks":1,"open_issues":0,"watchers":0,"default_branch":"master"}},"base":{"label":"apache:master","ref":"master","sha":"b0753f193cb7c8448547e254326911166d7b96a2","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":20587599,"name":"flink","full_name":"apache/flink","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/flink","description":"Mirror of Apache Flink","fork":false,"url":"https://api.github.com/repos/apache/flink","forks_url":"https://api.github.com/repos/apache/flink/forks","keys_url":"https://api.github.com/repos/apache/flink/keys{/key_id}","collaborators_url":"https://api.github.com/repos/apache/flink/collaborators{/collaborator}","teams_url":"https://api.github.com/repos/apache/flink/teams","hooks_url":"https://api.github.com/repos/apache/flink/hooks","issue_events_url":"https://api.github.com/repos/apache/flink/issues/events{/number}","events_url":"https://api.github.com/repos/apache/flink/events","assignees_url":"https://api.github.com/repos/apache/flink/assignees{/user}","branches_url":"https://api.github.com/repos/apache/flink/branches{/branch}","tags_url":"https://api.github.com/repos/apache/flink/tags","blobs_url":"https://api.github.com/repos/apache/flink/git/blobs{/sha}","git_tags_url":"https://api.github.com/repos/apache/flink/git/tags{/sha}","git_refs_url":"https://api.github.com/repos/apache/flink/git/refs{/sha}","trees_url":"https://api.github.com/repos/apache/flink/git/trees{/sha}","statuses_url":"https://api.github.com/repos/apache/flink/statuses/{sha}","languages_url":"https://api.github.com/repos/apache/flink/languages","stargazers_url":"https://api.github.com/repos/apache/flink/stargazers","contributors_url":"https://api.github.com/repos/apache/flink/contributors","subscribers_url":"https://api.github.com/repos/apache/flink/subscribers","subscription_url":"https://api.github.com/repos/apache/flink/subscription","commits_url":"https://api.github.com/repos/apache/flink/commits{/sha}","git_commits_url":"https://api.github.com/repos/apache/flink/git/commits{/sha}","comments_url":"https://api.github.com/repos/apache/flink/comments{/number}","issue_comment_url":"https://api.github.com/repos/apache/flink/issues/comments{/number}","contents_url":"https://api.github.com/repos/apache/flink/contents/{+path}","compare_url":"https://api.github.com/repos/apache/flink/compare/{base}...{head}","merges_url":"https://api.github.com/repos/apache/flink/merges","archive_url":"https://api.github.com/repos/apache/flink/{archive_format}{/ref}","downloads_url":"https://api.github.com/repos/apache/flink/downloads","issues_url":"https://api.github.com/repos/apache/flink/issues{/number}","pulls_url":"https://api.github.com/repos/apache/flink/pulls{/number}","milestones_url":"https://api.github.com/repos/apache/flink/milestones{/number}","notifications_url":"https://api.github.com/repos/apache/flink/notifications{?since,all,participating}","labels_url":"https://api.github.com/repos/apache/flink/labels{/name}","releases_url":"https://api.github.com/repos/apache/flink/releases{/id}","deployments_url":"https://api.github.com/repos/apache/flink/deployments","created_at":"2014-06-07T07:00:10Z","updated_at":"2016-11-23T14:17:53Z","pushed_at":"2016-11-23T16:58:44Z","git_url":"git://github.com/apache/flink.git","ssh_url":"git@github.com:apache/flink.git","clone_url":"https://github.com/apache/flink.git","svn_url":"https://github.com/apache/flink","homepage":null,"size":143412,"stargazers_count":1770,"watchers_count":1770,"language":"Java","has_issues":false,"has_downloads":true,"has_wiki":false,"has_pages":false,"forks_count":1199,"mirror_url":"git://git.apache.org/flink.git","open_issues_count":186,"forks":1199,"open_issues":186,"watchers":1770,"default_branch":"master"}},"_links":{"self":{"href":"https://api.github.com/repos/apache/flink/pulls/2509"},"html":{"href":"https://github.com/apache/flink/pull/2509"},"issue":{"href":"https://api.github.com/repos/apache/flink/issues/2509"},"comments":{"href":"https://api.github.com/repos/apache/flink/issues/2509/comments"},"review_comments":{"href":"https://api.github.com/repos/apache/flink/pulls/2509/comments"},"review_comment":{"href":"https://api.github.com/repos/apache/flink/pulls/comments{/number}"},"commits":{"href":"https://api.github.com/repos/apache/flink/pulls/2509/commits"},"statuses":{"href":"https://api.github.com/repos/apache/flink/statuses/d8f5f976ef2e2e1d994a45468d7e9ef3b8bf0015"}}}}
{"actor":{"display_login":"tzulitai"}}