{"action":"created","comment":{"url":"https://api.github.com/repos/apache/flink/pulls/comments/69274851","id":69274851,"diff_hunk":"@@ -60,10 +60,10 @@ to setup Kinesis streams. Make sure to create the appropriate IAM policy and use\n \n ### Kinesis Consumer","path":"docs/apis/streaming/connectors/kinesis.md","position":2,"original_position":2,"commit_id":"b1f1e6fba5ee0990e6c30ce6628f97811cb465bd","original_commit_id":"b1f1e6fba5ee0990e6c30ce6628f97811cb465bd","user":{"login":"rmetzger","id":89049,"avatar_url":"https://avatars.githubusercontent.com/u/89049?v=3","gravatar_id":"","url":"https://api.github.com/users/rmetzger","html_url":"https://github.com/rmetzger","followers_url":"https://api.github.com/users/rmetzger/followers","following_url":"https://api.github.com/users/rmetzger/following{/other_user}","gists_url":"https://api.github.com/users/rmetzger/gists{/gist_id}","starred_url":"https://api.github.com/users/rmetzger/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/rmetzger/subscriptions","organizations_url":"https://api.github.com/users/rmetzger/orgs","repos_url":"https://api.github.com/users/rmetzger/repos","events_url":"https://api.github.com/users/rmetzger/events{/privacy}","received_events_url":"https://api.github.com/users/rmetzger/received_events","type":"User","site_admin":false},"body":"I think we should also mention the threading model of the Kinesis consumer in the documentation.\r\n\r\nUsers should know that each parallel Flink instance will constantly query Kinesis for shards. So if a user has 5 shards, but runs Flink with a parallelism of 50, there will be 50 threads constantly querying Kinesis.\r\n\r\nAlso, we should explain that there will always be one thread per shard running.","created_at":"2016-07-01T09:44:24Z","updated_at":"2016-07-01T09:44:24Z","html_url":"https://github.com/apache/flink/pull/2131#discussion_r69274851","pull_request_url":"https://api.github.com/repos/apache/flink/pulls/2131","_links":{"self":{"href":"https://api.github.com/repos/apache/flink/pulls/comments/69274851"},"html":{"href":"https://github.com/apache/flink/pull/2131#discussion_r69274851"},"pull_request":{"href":"https://api.github.com/repos/apache/flink/pulls/2131"}}},"pull_request":{"url":"https://api.github.com/repos/apache/flink/pulls/2131","id":74368601,"html_url":"https://github.com/apache/flink/pull/2131","diff_url":"https://github.com/apache/flink/pull/2131.diff","patch_url":"https://github.com/apache/flink/pull/2131.patch","issue_url":"https://api.github.com/repos/apache/flink/issues/2131","number":2131,"state":"open","locked":false,"title":"[FLINK-3231][streaming-connectors] FlinkKinesisConsumer rework to handle Kinesis resharding","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 change attempts to solve 2 issues:\r\n1. [FLINK-3231](https://issues.apache.org/jira/browse/FLINK-3231): Handle Kinesis-side resharding.\r\n2. [FLINK-4020](https://issues.apache.org/jira/browse/FLINK-4020): Remove shard list querying from Kinesis consumer constructor.\r\n\r\nSome notes on the implementation:\r\n- All subtasks has a thread that continuously polls for changes in the Kinesis stream, and uses exponential backoff with jitter to try to even out the concurrent Kinesis client describeStream operations across subtasks. Continuously polling is necessary because there's currently no way to \"signal\" a subtask that it has a new shard it should be subscribing to.\r\n- A big change is that all subtasks run a fetcher that continues to poll for shards, even if the subtask initially didn't have shards to consume (before, a MAX_VALUE waterwark was sent out).\r\n- The `describeStream` operation doesn't fail after 3 retries now. Instead, if the failures it was getting from Kinesis was only `LimitExceededException`, we continue to retry (with jitter backoff) until we succeed.\r\n- Apart from the unit tests, I've manually tested this with `ManualExactlyOnceWithStreamReshardingTest`. However, since the `FlinkKinesisProducer` currently has some problem that records are resent when Kinesis streams are resharded (thus not allowing the exactly-once test to pass at all), this manual test uses a normal event generator instead of a producer topology running the `FlinkKinesisProducer`.\r\n\r\nSince this PR introduces considerable rework on the Kinesis consumer, I'll wait until this can be merged before submitting [FLINK-4080](https://issues.apache.org/jira/browse/FLINK-4080) & [FLINK-4019](https://issues.apache.org/jira/browse/FLINK-4019).","created_at":"2016-06-19T17:15:29Z","updated_at":"2016-07-01T09:44:24Z","closed_at":null,"merged_at":null,"merge_commit_sha":"6babfd868b66fb35e0cf02723bc16c4c808fc352","assignee":null,"assignees":[],"milestone":null,"commits_url":"https://api.github.com/repos/apache/flink/pulls/2131/commits","review_comments_url":"https://api.github.com/repos/apache/flink/pulls/2131/comments","review_comment_url":"https://api.github.com/repos/apache/flink/pulls/comments{/number}","comments_url":"https://api.github.com/repos/apache/flink/issues/2131/comments","statuses_url":"https://api.github.com/repos/apache/flink/statuses/b1f1e6fba5ee0990e6c30ce6628f97811cb465bd","head":{"label":"tzulitai:FLINK-3231","ref":"FLINK-3231","sha":"b1f1e6fba5ee0990e6c30ce6628f97811cb465bd","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-07-01T08:57:28Z","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":91782,"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":"53630da01bcbfe05eda90869b1198b4e1c554a86","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-07-01T06:58:10Z","pushed_at":"2016-07-01T09:04:22Z","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":120879,"stargazers_count":1399,"watchers_count":1399,"language":"Java","has_issues":false,"has_downloads":true,"has_wiki":false,"has_pages":false,"forks_count":904,"mirror_url":"git://git.apache.org/flink.git","open_issues_count":119,"forks":904,"open_issues":119,"watchers":1399,"default_branch":"master"}},"_links":{"self":{"href":"https://api.github.com/repos/apache/flink/pulls/2131"},"html":{"href":"https://github.com/apache/flink/pull/2131"},"issue":{"href":"https://api.github.com/repos/apache/flink/issues/2131"},"comments":{"href":"https://api.github.com/repos/apache/flink/issues/2131/comments"},"review_comments":{"href":"https://api.github.com/repos/apache/flink/pulls/2131/comments"},"review_comment":{"href":"https://api.github.com/repos/apache/flink/pulls/comments{/number}"},"commits":{"href":"https://api.github.com/repos/apache/flink/pulls/2131/commits"},"statuses":{"href":"https://api.github.com/repos/apache/flink/statuses/b1f1e6fba5ee0990e6c30ce6628f97811cb465bd"}}}}
{"actor":{"display_login":"rmetzger"}}