{"action":"created","comment":{"url":"https://api.github.com/repos/apache/flink/pulls/comments/114795536","pull_request_review_id":36293088,"id":114795536,"diff_hunk":"@@ -219,6 +209,15 @@ public void run() {\n \t\t\t\t\t}\n \t\t\t\t}\n \n+\t\t\t\ttry {\n+\t\t\t\t\tnewPartitions = unassignedPartitionsQueue.pollBatch();\n+\t\t\t\t\tif (newPartitions != null) {\n+\t\t\t\t\t\treassignPartitions(newPartitions);\n+\t\t\t\t\t}\n+\t\t\t\t} catch (AbortedReassignmentException e) {\n+\t\t\t\t\tcontinue;","path":"flink-connectors/flink-connector-kafka-0.9/src/main/java/org/apache/flink/streaming/connectors/kafka/internal/KafkaConsumerThread.java","position":148,"original_position":148,"commit_id":"bf2dd78e7706b9aae08b75ff917878e3f2ceb68b","original_commit_id":"bf2dd78e7706b9aae08b75ff917878e3f2ceb68b","user":{"login":"tzulitai","id":5284370,"avatar_url":"https://avatars0.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":"the reason is that the `AbortedReassignmentException` is thrown only if the Kafka consumer was `wakeup` before the reassignment started and that wakeup effect was consumed by an operation during the reassignment.\r\n\r\nNow, if the loop just continues, the next Kafka operation (the `poll`) would not be woken up because the wakeup was already consumed.\r\nSo, effectively, it would mean the same thing to do this in the catch block instead of falling through the loop:\r\n```\r\n} catch (AbortedReassignmentException e) {\r\nconsumer.wakeup();\r\n}\r\n```\r\n\r\nI'm just eagerly continuing instead of restoring the wakeup. Does that make sense?","created_at":"2017-05-04T14:56:10Z","updated_at":"2017-05-04T14:56:10Z","html_url":"https://github.com/apache/flink/pull/3746#discussion_r114795536","pull_request_url":"https://api.github.com/repos/apache/flink/pulls/3746","_links":{"self":{"href":"https://api.github.com/repos/apache/flink/pulls/comments/114795536"},"html":{"href":"https://github.com/apache/flink/pull/3746#discussion_r114795536"},"pull_request":{"href":"https://api.github.com/repos/apache/flink/pulls/3746"}}},"pull_request":{"url":"https://api.github.com/repos/apache/flink/pulls/3746","id":116799032,"html_url":"https://github.com/apache/flink/pull/3746","diff_url":"https://github.com/apache/flink/pull/3746.diff","patch_url":"https://github.com/apache/flink/pull/3746.patch","issue_url":"https://api.github.com/repos/apache/flink/issues/3746","number":3746,"state":"open","locked":false,"title":"[FLINK-4022] [kafka] Partition and topic discovery for FlinkKafkaConsumer","user":{"login":"tzulitai","id":5284370,"avatar_url":"https://avatars0.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 PR adds the required internals to allow partition and topic regex pattern discovery in the `FlinkKafkaConsumer`.\r\n\r\nIt doesn't expose a new constructor that accepts regex topic patterns yet. I propose to expose that with https://issues.apache.org/jira/browse/FLINK-5704 (deprecate the original FlinkKafkaConsumer constructors in favor of new ones with new offset behaviours). For this reason, I also propose to update the Kafka documentation when the new constructors are added.\r\n\r\n## Design\r\n\r\nSome description to ease review:\r\n\r\n- `AbstractPartitionDiscoverer`:\r\nAn `AbstractPartitionDiscoverer` is a stateful utility instance that remembers what partitions are discovered already. It also wraps the logic for partition-to-subtask assignment. The main `run()` method now has a discovery loop that calls `AbstractPartitionDiscoverer#discoverPartitions()` on a fixed interval. This method returns only new partitions that should be subscribed by the subtask.\r\nThe returned partitions are used to invoke `AbstractFetcher#addDiscoveredPartitions(...)` on the fetcher.\r\nOn a fresh startup, `AbstractPartitionDiscoverer#discoverPartitions()` is also used to fetch the initial seed startup partitions in `open()`.\r\n\r\n- `AbstractFetcher#addDiscoveredPartitions(...)`\r\nThe fetcher now has an `unassignedPartitionsQueue` that contains discovered partitions not yet assigned to Kafka consumers to be read. Whenever `addDiscoveredPartitions(...)` is called on the fetcher, the fetcher will create the state holders for the partitions, and add the partitions to the queue.\r\nConcrete implementations of the fetcher should continuously poll this queue in the fetch loop. If partitions are found from the queue, they should be assigned for consuming.\r\n\r\n- Concrete fetchers continuously polls the queue in `runFetchLoop()`\r\nFor 0.8, this simply means that the original `unassignedPartitionsQueue` in Kafka08Fetcher is moved to the base abstract fetcher class. Nothing else is touched.\r\nFor 0.9+, queue polling and partition reassignment for the high-level consumer happens in `KafkaConsumerThread`.\r\n\r\nWith this design, partition discovery is split into 2 separate concerns:\r\n- Fetchers simply polls the queue for new partitions to read from in their fetch loop.\r\n- The main run loop serves as the discovery loop. Whenever new partitions are found, they are added to the queue. Initial partitions that are found on fresh startups (retrieved before entering the discovery loop) are treated as seeded elements in the queue.\r\n\r\n## Limitations\r\n\r\nFor the partition discovery to work properly after restores, the previous state had to be migrated to use union list states instead (`OperatorStateStore#getUnionListState()`).\r\n\r\nOne limitation that comes with this migration is that if the current run was restored from old state (Flink 1.1 / 1.2), then partition discovery is disabled. In order to use partition discovery, after the job starts the user must do a manual snapshot (with the new union state) & restore.","created_at":"2017-04-20T15:26:34Z","updated_at":"2017-05-04T14:56:10Z","closed_at":null,"merged_at":null,"merge_commit_sha":"584e9874f334d574fa5264171b8f46cda93144a7","assignee":null,"assignees":[],"milestone":null,"commits_url":"https://api.github.com/repos/apache/flink/pulls/3746/commits","review_comments_url":"https://api.github.com/repos/apache/flink/pulls/3746/comments","review_comment_url":"https://api.github.com/repos/apache/flink/pulls/comments{/number}","comments_url":"https://api.github.com/repos/apache/flink/issues/3746/comments","statuses_url":"https://api.github.com/repos/apache/flink/statuses/bf2dd78e7706b9aae08b75ff917878e3f2ceb68b","head":{"label":"tzulitai:FLINK-4022","ref":"FLINK-4022","sha":"bf2dd78e7706b9aae08b75ff917878e3f2ceb68b","user":{"login":"tzulitai","id":5284370,"avatar_url":"https://avatars0.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://avatars0.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":"2017-05-04T08:14:00Z","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":120285,"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":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":"42328bd9b7f216e4c3aae2086b822b4a3a564970","user":{"login":"apache","id":47359,"avatar_url":"https://avatars3.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://avatars3.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":"2017-05-04T08:19:58Z","pushed_at":"2017-05-04T13:58:59Z","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":120354,"stargazers_count":2273,"watchers_count":2273,"language":"Java","has_issues":false,"has_projects":true,"has_downloads":true,"has_wiki":false,"has_pages":false,"forks_count":1600,"mirror_url":"git://git.apache.org/flink.git","open_issues_count":243,"forks":1600,"open_issues":243,"watchers":2273,"default_branch":"master"}},"_links":{"self":{"href":"https://api.github.com/repos/apache/flink/pulls/3746"},"html":{"href":"https://github.com/apache/flink/pull/3746"},"issue":{"href":"https://api.github.com/repos/apache/flink/issues/3746"},"comments":{"href":"https://api.github.com/repos/apache/flink/issues/3746/comments"},"review_comments":{"href":"https://api.github.com/repos/apache/flink/pulls/3746/comments"},"review_comment":{"href":"https://api.github.com/repos/apache/flink/pulls/comments{/number}"},"commits":{"href":"https://api.github.com/repos/apache/flink/pulls/3746/commits"},"statuses":{"href":"https://api.github.com/repos/apache/flink/statuses/bf2dd78e7706b9aae08b75ff917878e3f2ceb68b"}}}}
{"actor":{"display_login":"tzulitai"}}