{"action":"created","comment":{"url":"https://api.github.com/repos/apache/flink/pulls/comments/189164871","pull_request_review_id":121289628,"id":189164871,"diff_hunk":"@@ -326,6 +342,24 @@ private void checkAndPropagateAsyncError() throws Exception {\n \t\t}\n \t}\n \n+\t/**\n+\t * If the internal queue of the {@link KinesisProducer} gets too long,\n+\t * flush some of the records until we are below the limit again.\n+\t * We don't want to flush _all_ records at this point since that would\n+\t * break record aggregation.\n+\t */\n+\tprivate void checkQueueLimit() {\n+\t\twhile (producer.getOutstandingRecordsCount() >= queueLimit) {\n+\t\t\tproducer.flush();","path":"flink-connectors/flink-connector-kinesis/src/main/java/org/apache/flink/streaming/connectors/kinesis/FlinkKinesisProducer.java","position":49,"original_position":49,"commit_id":"690a146e2933f60d4dc6d11c59bcafdbc881faeb","original_commit_id":"690a146e2933f60d4dc6d11c59bcafdbc881faeb","user":{"login":"tzulitai","id":5284370,"avatar_url":"https://avatars3.githubusercontent.com/u/5284370?v=4","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":"Do we have to do a flush here? Shouldn't the KPL child process process the user records in the background without an explicit flush call?\r\n\r\nIf so, perhaps a more graceful solution here is to wait on a local object, and notify it to wake up in the asynchronous producer write call backs. After being notified, we check the `getOutstandingRecordsCount` agains the queueLimit, and either wait more or escape the loop.\r\n\r\nWhat do you think?","created_at":"2018-05-18T05:13:52Z","updated_at":"2018-05-18T05:17:54Z","html_url":"https://github.com/apache/flink/pull/6021#discussion_r189164871","pull_request_url":"https://api.github.com/repos/apache/flink/pulls/6021","author_association":"CONTRIBUTOR","_links":{"self":{"href":"https://api.github.com/repos/apache/flink/pulls/comments/189164871"},"html":{"href":"https://github.com/apache/flink/pull/6021#discussion_r189164871"},"pull_request":{"href":"https://api.github.com/repos/apache/flink/pulls/6021"}}},"pull_request":{"url":"https://api.github.com/repos/apache/flink/pulls/6021","id":188350607,"html_url":"https://github.com/apache/flink/pull/6021","diff_url":"https://github.com/apache/flink/pull/6021.diff","patch_url":"https://github.com/apache/flink/pull/6021.patch","issue_url":"https://api.github.com/repos/apache/flink/issues/6021","number":6021,"state":"open","locked":false,"title":"[FLINK-9374] [kinesis] Enable FlinkKinesisProducer Backpressuring","user":{"login":"fmthoma","id":5918766,"avatar_url":"https://avatars2.githubusercontent.com/u/5918766?v=4","gravatar_id":"","url":"https://api.github.com/users/fmthoma","html_url":"https://github.com/fmthoma","followers_url":"https://api.github.com/users/fmthoma/followers","following_url":"https://api.github.com/users/fmthoma/following{/other_user}","gists_url":"https://api.github.com/users/fmthoma/gists{/gist_id}","starred_url":"https://api.github.com/users/fmthoma/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/fmthoma/subscriptions","organizations_url":"https://api.github.com/users/fmthoma/orgs","repos_url":"https://api.github.com/users/fmthoma/repos","events_url":"https://api.github.com/users/fmthoma/events{/privacy}","received_events_url":"https://api.github.com/users/fmthoma/received_events","type":"User","site_admin":false},"body":"## What is the purpose of the change\r\n\r\nThe `FlinkKinesisProducer` just accepts records and forwards it to a `KinesisProducer` from the Amazon Kinesis Producer Library (KPL). The KPL internally holds an unbounded queue of records that have not yet been sent.\r\n\r\nSince Kinesis is rate-limited to 1MB per second per shard, this queue may grow indefinitely if Flink sends records faster than the KPL can forward them to Kinesis.\r\n\r\nOne way to circumvent this problem is to set a record TTL, so that queued records are dropped after a certain amount of time, but this will lead to data loss under high loads.\r\n\r\nCurrently the only time the queue is flushed is during checkpointing: `FlinkKinesisProducer` consumes records at arbitrary rate, either until a checkpoint is reached (and will wait until the queue is flushed), or until out-of-memory, whichever is reached first. (This gets worse due to the fact that the Java KPL is only a thin wrapper around a C++ process, so it is not even the Java process that runs out of memory, but the C++ process.)\r\n\r\nMy proposed solution is to add a config option `queueLimit` to set a maximum number of records that may be waiting in the KPL queue. If this limit is reached, the `FlinkKinesisProducer` should trigger a `flush()` and wait (blocking) until the queue length is below the limit again. This automatically leads to backpressuring, since the `FlinkKinesisProducer` cannot accept records while waiting. For compatibility, `queueLimit` is set to `Integer.MAX_VALUE` by default, so the behavior is unchanged unless a client explicitly sets the value. Setting a »sane« default value is not possible unfortunately, since sensible values for the limit depend on the record size (the limit should be chosen so that about 10–100MB of records per shard are accumulated before flushing, otherwise the maximum Kinesis throughput may not be reached).\r\n\r\n## Brief change log\r\n\r\n* Add a `queueLimit` setting to `FlinkKinesisProducer` to limit the number of in-flight records in the Kinesis Producer Library, and enable backpressuring if the limit is exceeded\r\n\r\n## Verifying this change\r\n\r\nThis change added tests and can be verified as follows:\r\n\r\n* Added unit test\r\n* Manually verified the change by running a job that produces to a 2-shard Kinesis stream. The input rate is limited by Kinesis (verified that the Kinesis stream is indeed at maximum capacity).\r\n\r\n## Does this pull request potentially affect one of the following parts:\r\n\r\n  - Dependencies (does it add or upgrade a dependency): no\r\n  - The public API, i.e., is any changed class annotated with `@Public(Evolving)`: yes, but backwards compatible (option was added)\r\n  - The serializers: no\r\n  - The runtime per-record code paths (performance sensitive): don't know\r\n  - Anything that affects deployment or recovery: JobManager (and its components), Checkpointing, Yarn/Mesos, ZooKeeper: don't know\r\n  - The S3 file system connector: no\r\n\r\n## Documentation\r\n\r\n  - Does this pull request introduce a new feature? yes\r\n  - If yes, how is the feature documented? JavaDocs","created_at":"2018-05-16T08:12:14Z","updated_at":"2018-05-18T05:17:54Z","closed_at":null,"merged_at":null,"merge_commit_sha":"a73ce77463bc6c33fb168f2164a08ff158723803","assignee":null,"assignees":[],"requested_reviewers":[],"requested_teams":[],"labels":[],"milestone":null,"commits_url":"https://api.github.com/repos/apache/flink/pulls/6021/commits","review_comments_url":"https://api.github.com/repos/apache/flink/pulls/6021/comments","review_comment_url":"https://api.github.com/repos/apache/flink/pulls/comments{/number}","comments_url":"https://api.github.com/repos/apache/flink/issues/6021/comments","statuses_url":"https://api.github.com/repos/apache/flink/statuses/690a146e2933f60d4dc6d11c59bcafdbc881faeb","head":{"label":"fmthoma:queueLimit","ref":"queueLimit","sha":"690a146e2933f60d4dc6d11c59bcafdbc881faeb","user":{"login":"fmthoma","id":5918766,"avatar_url":"https://avatars2.githubusercontent.com/u/5918766?v=4","gravatar_id":"","url":"https://api.github.com/users/fmthoma","html_url":"https://github.com/fmthoma","followers_url":"https://api.github.com/users/fmthoma/followers","following_url":"https://api.github.com/users/fmthoma/following{/other_user}","gists_url":"https://api.github.com/users/fmthoma/gists{/gist_id}","starred_url":"https://api.github.com/users/fmthoma/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/fmthoma/subscriptions","organizations_url":"https://api.github.com/users/fmthoma/orgs","repos_url":"https://api.github.com/users/fmthoma/repos","events_url":"https://api.github.com/users/fmthoma/events{/privacy}","received_events_url":"https://api.github.com/users/fmthoma/received_events","type":"User","site_admin":false},"repo":{"id":132614305,"name":"flink","full_name":"fmthoma/flink","owner":{"login":"fmthoma","id":5918766,"avatar_url":"https://avatars2.githubusercontent.com/u/5918766?v=4","gravatar_id":"","url":"https://api.github.com/users/fmthoma","html_url":"https://github.com/fmthoma","followers_url":"https://api.github.com/users/fmthoma/followers","following_url":"https://api.github.com/users/fmthoma/following{/other_user}","gists_url":"https://api.github.com/users/fmthoma/gists{/gist_id}","starred_url":"https://api.github.com/users/fmthoma/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/fmthoma/subscriptions","organizations_url":"https://api.github.com/users/fmthoma/orgs","repos_url":"https://api.github.com/users/fmthoma/repos","events_url":"https://api.github.com/users/fmthoma/events{/privacy}","received_events_url":"https://api.github.com/users/fmthoma/received_events","type":"User","site_admin":false},"private":false,"html_url":"https://github.com/fmthoma/flink","description":"Mirror of Apache Flink","fork":true,"url":"https://api.github.com/repos/fmthoma/flink","forks_url":"https://api.github.com/repos/fmthoma/flink/forks","keys_url":"https://api.github.com/repos/fmthoma/flink/keys{/key_id}","collaborators_url":"https://api.github.com/repos/fmthoma/flink/collaborators{/collaborator}","teams_url":"https://api.github.com/repos/fmthoma/flink/teams","hooks_url":"https://api.github.com/repos/fmthoma/flink/hooks","issue_events_url":"https://api.github.com/repos/fmthoma/flink/issues/events{/number}","events_url":"https://api.github.com/repos/fmthoma/flink/events","assignees_url":"https://api.github.com/repos/fmthoma/flink/assignees{/user}","branches_url":"https://api.github.com/repos/fmthoma/flink/branches{/branch}","tags_url":"https://api.github.com/repos/fmthoma/flink/tags","blobs_url":"https://api.github.com/repos/fmthoma/flink/git/blobs{/sha}","git_tags_url":"https://api.github.com/repos/fmthoma/flink/git/tags{/sha}","git_refs_url":"https://api.github.com/repos/fmthoma/flink/git/refs{/sha}","trees_url":"https://api.github.com/repos/fmthoma/flink/git/trees{/sha}","statuses_url":"https://api.github.com/repos/fmthoma/flink/statuses/{sha}","languages_url":"https://api.github.com/repos/fmthoma/flink/languages","stargazers_url":"https://api.github.com/repos/fmthoma/flink/stargazers","contributors_url":"https://api.github.com/repos/fmthoma/flink/contributors","subscribers_url":"https://api.github.com/repos/fmthoma/flink/subscribers","subscription_url":"https://api.github.com/repos/fmthoma/flink/subscription","commits_url":"https://api.github.com/repos/fmthoma/flink/commits{/sha}","git_commits_url":"https://api.github.com/repos/fmthoma/flink/git/commits{/sha}","comments_url":"https://api.github.com/repos/fmthoma/flink/comments{/number}","issue_comment_url":"https://api.github.com/repos/fmthoma/flink/issues/comments{/number}","contents_url":"https://api.github.com/repos/fmthoma/flink/contents/{+path}","compare_url":"https://api.github.com/repos/fmthoma/flink/compare/{base}...{head}","merges_url":"https://api.github.com/repos/fmthoma/flink/merges","archive_url":"https://api.github.com/repos/fmthoma/flink/{archive_format}{/ref}","downloads_url":"https://api.github.com/repos/fmthoma/flink/downloads","issues_url":"https://api.github.com/repos/fmthoma/flink/issues{/number}","pulls_url":"https://api.github.com/repos/fmthoma/flink/pulls{/number}","milestones_url":"https://api.github.com/repos/fmthoma/flink/milestones{/number}","notifications_url":"https://api.github.com/repos/fmthoma/flink/notifications{?since,all,participating}","labels_url":"https://api.github.com/repos/fmthoma/flink/labels{/name}","releases_url":"https://api.github.com/repos/fmthoma/flink/releases{/id}","deployments_url":"https://api.github.com/repos/fmthoma/flink/deployments","created_at":"2018-05-08T13:38:19Z","updated_at":"2018-05-08T13:38:48Z","pushed_at":"2018-05-16T11:52:49Z","git_url":"git://github.com/fmthoma/flink.git","ssh_url":"git@github.com:fmthoma/flink.git","clone_url":"https://github.com/fmthoma/flink.git","svn_url":"https://github.com/fmthoma/flink","homepage":null,"size":154568,"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,"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"},"forks":0,"open_issues":0,"watchers":0,"default_branch":"master"}},"base":{"label":"apache:master","ref":"master","sha":"0b5e124ba24230026995cbb93444f30af7d51363","user":{"login":"apache","id":47359,"avatar_url":"https://avatars0.githubusercontent.com/u/47359?v=4","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://avatars0.githubusercontent.com/u/47359?v=4","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":"2018-05-17T23:05:09Z","pushed_at":"2018-05-18T05:13:41Z","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":155000,"stargazers_count":3704,"watchers_count":3704,"language":"Java","has_issues":false,"has_projects":true,"has_downloads":true,"has_wiki":false,"has_pages":false,"forks_count":2519,"mirror_url":null,"archived":false,"open_issues_count":352,"license":{"key":"apache-2.0","name":"Apache License 2.0","spdx_id":"Apache-2.0","url":"https://api.github.com/licenses/apache-2.0"},"forks":2519,"open_issues":352,"watchers":3704,"default_branch":"master"}},"_links":{"self":{"href":"https://api.github.com/repos/apache/flink/pulls/6021"},"html":{"href":"https://github.com/apache/flink/pull/6021"},"issue":{"href":"https://api.github.com/repos/apache/flink/issues/6021"},"comments":{"href":"https://api.github.com/repos/apache/flink/issues/6021/comments"},"review_comments":{"href":"https://api.github.com/repos/apache/flink/pulls/6021/comments"},"review_comment":{"href":"https://api.github.com/repos/apache/flink/pulls/comments{/number}"},"commits":{"href":"https://api.github.com/repos/apache/flink/pulls/6021/commits"},"statuses":{"href":"https://api.github.com/repos/apache/flink/statuses/690a146e2933f60d4dc6d11c59bcafdbc881faeb"}},"author_association":"NONE"}}
{"actor":{"display_login":"tzulitai"}}