{"action":"created","comment":{"url":"https://api.github.com/repos/confluentinc/ksql/pulls/comments/216426809","pull_request_review_id":153904811,"id":216426809,"node_id":"MDI0OlB1bGxSZXF1ZXN0UmV2aWV3Q29tbWVudDIxNjQyNjgwOQ==","diff_hunk":"@@ -131,6 +140,14 @@ void handleStatement(\n     return Optional.ofNullable(statusStore.get(statementId));\n   }\n \n+  public void putStatus(final CommandId commandId, final Optional<CommandStatusFuture> statusFuture,\n+                        final CommandStatus status) {\n+    statusStore.put(commandId, status);\n+    if (statusFuture.isPresent()) {","path":"ksql-rest-app/src/main/java/io/confluent/ksql/rest/server/computation/StatementExecutor.java","position":80,"original_position":80,"commit_id":"a9cbdc49afce2601b71687d2717f19e8ecab2a1c","original_commit_id":"a9cbdc49afce2601b71687d2717f19e8ecab2a1c","user":{"login":"rodesai","id":2610028,"node_id":"MDQ6VXNlcjI2MTAwMjg=","avatar_url":"https://avatars0.githubusercontent.com/u/2610028?v=4","gravatar_id":"","url":"https://api.github.com/users/rodesai","html_url":"https://github.com/rodesai","followers_url":"https://api.github.com/users/rodesai/followers","following_url":"https://api.github.com/users/rodesai/following{/other_user}","gists_url":"https://api.github.com/users/rodesai/gists{/gist_id}","starred_url":"https://api.github.com/users/rodesai/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/rodesai/subscriptions","organizations_url":"https://api.github.com/users/rodesai/orgs","repos_url":"https://api.github.com/users/rodesai/repos","events_url":"https://api.github.com/users/rodesai/events{/privacy}","received_events_url":"https://api.github.com/users/rodesai/received_events","type":"User","site_admin":false},"body":"The idea here is to provide an object that StatementExecutor can use to communicate command processing status updates to KsqlResource, and that KsqlResource can block on to be notified by StatementExecutor when its done with a command. I think maybe keeping Future in the name is making it confusing, since it actually goes through multiple states, but we only want to mark it \"complete\" once StatementExecutor is finished. We could also decouple tracking the shared state object and wait/notify stuff.\r\n\r\nAlso, I'm not really sure what the point of the intermediate states is. If we took them out we could simplify this stuff a bit. I didn't go that route because:\r\n    1 - I'm not totally sure its not used somewhere.\r\n    2 - That changes the API and I wanted to keep this PR as concise as possible since its fixing a test stability problem.","created_at":"2018-09-10T18:27:58Z","updated_at":"2018-09-10T18:27:58Z","html_url":"https://github.com/confluentinc/ksql/pull/1856#discussion_r216426809","pull_request_url":"https://api.github.com/repos/confluentinc/ksql/pulls/1856","author_association":"CONTRIBUTOR","_links":{"self":{"href":"https://api.github.com/repos/confluentinc/ksql/pulls/comments/216426809"},"html":{"href":"https://github.com/confluentinc/ksql/pull/1856#discussion_r216426809"},"pull_request":{"href":"https://api.github.com/repos/confluentinc/ksql/pulls/1856"}},"in_reply_to_id":216417537},"pull_request":{"url":"https://api.github.com/repos/confluentinc/ksql/pulls/1856","id":214267335,"node_id":"MDExOlB1bGxSZXF1ZXN0MjE0MjY3MzM1","html_url":"https://github.com/confluentinc/ksql/pull/1856","diff_url":"https://github.com/confluentinc/ksql/pull/1856.diff","patch_url":"https://github.com/confluentinc/ksql/pull/1856.patch","issue_url":"https://api.github.com/repos/confluentinc/ksql/issues/1856","number":1856,"state":"open","locked":false,"title":"Fix some synchronization problems with KsqlResource and StatementExec…","user":{"login":"rodesai","id":2610028,"node_id":"MDQ6VXNlcjI2MTAwMjg=","avatar_url":"https://avatars0.githubusercontent.com/u/2610028?v=4","gravatar_id":"","url":"https://api.github.com/users/rodesai","html_url":"https://github.com/rodesai","followers_url":"https://api.github.com/users/rodesai/followers","following_url":"https://api.github.com/users/rodesai/following{/other_user}","gists_url":"https://api.github.com/users/rodesai/gists{/gist_id}","starred_url":"https://api.github.com/users/rodesai/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/rodesai/subscriptions","organizations_url":"https://api.github.com/users/rodesai/orgs","repos_url":"https://api.github.com/users/rodesai/repos","events_url":"https://api.github.com/users/rodesai/events{/privacy}","received_events_url":"https://api.github.com/users/rodesai/received_events","type":"User","site_admin":false},"body":"…utor\r\n\r\nKsqlResource and StatementExecutor are not synchronized properly:\r\n    - KsqlResource registers for a future after writing commands to the\r\n      command topic. By this time, the StatementExecutor may have already\r\n      picked up the command and missed the future created by KsqlResource\r\n    - If multiple threads try to perform the same operation around the same\r\n      time, they could both end up listening on the same future and one of\r\n      them will then return an incorrect response. This can also happen\r\n      across servers, but this PR does not attempt to solve that (will file\r\n      a follow-up to cover that case, since its a bit more controvertial\r\n      as it requires a change to the command messages).\r\n    - The id -> status map is not synchronized, but is accessed by multiple\r\n      threads\r\n\r\nTo fix these problems, we've done the following:\r\n\r\nKsqlResource/CommandStore now register for futures before writing to the\r\ncommand topic. All communication between the StatementExecutor and\r\nKsqlResource happens over this future.\r\n\r\nIf another request comes in and tries to register for a future for the\r\nsame logical operation, we fail the request. As stated above, this could\r\nalso happen across servers, in which case the implemented check won't help.\r\nTo fix that, we'll need to add a \"cookie\" to the command topic that\r\nStatementExecutor can look up to find the future that KsqlResource is waiting\r\non. This way we can make a strong correlation between a command being executed\r\nand a waiting server thread. That change will come in a follow-on PR (at which\r\npoint we could get rid of the check).\r\n\r\nUse a ConcurrentHashMap for the id -> status map.\r\n\r\n### Description \r\n_What behavior do you want to change, why, how does your patch achieve the changes?_\r\n\r\n### Testing done \r\n_Describe the testing stategy. Unit and integration tests are expected for any behavior changes._\r\n\r\n### Reviewer checklist\r\n- [ ] Ensure docs are updated if necessary. (eg. if a user visible feature is being added or changed).\r\n- [ ] Ensure relevant issues are linked (description should include text like \"Fixes #<issue number>\")\r\n\r\n","created_at":"2018-09-10T10:29:16Z","updated_at":"2018-09-10T18:27:58Z","closed_at":null,"merged_at":null,"merge_commit_sha":"f65a04c3220329af4dcdfd7c8d9110b3d7b9ea6c","assignee":null,"assignees":[],"requested_reviewers":[],"requested_teams":[],"labels":[],"milestone":null,"commits_url":"https://api.github.com/repos/confluentinc/ksql/pulls/1856/commits","review_comments_url":"https://api.github.com/repos/confluentinc/ksql/pulls/1856/comments","review_comment_url":"https://api.github.com/repos/confluentinc/ksql/pulls/comments{/number}","comments_url":"https://api.github.com/repos/confluentinc/ksql/issues/1856/comments","statuses_url":"https://api.github.com/repos/confluentinc/ksql/statuses/a9cbdc49afce2601b71687d2717f19e8ecab2a1c","head":{"label":"rodesai:server_executor_synch","ref":"server_executor_synch","sha":"a9cbdc49afce2601b71687d2717f19e8ecab2a1c","user":{"login":"rodesai","id":2610028,"node_id":"MDQ6VXNlcjI2MTAwMjg=","avatar_url":"https://avatars0.githubusercontent.com/u/2610028?v=4","gravatar_id":"","url":"https://api.github.com/users/rodesai","html_url":"https://github.com/rodesai","followers_url":"https://api.github.com/users/rodesai/followers","following_url":"https://api.github.com/users/rodesai/following{/other_user}","gists_url":"https://api.github.com/users/rodesai/gists{/gist_id}","starred_url":"https://api.github.com/users/rodesai/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/rodesai/subscriptions","organizations_url":"https://api.github.com/users/rodesai/orgs","repos_url":"https://api.github.com/users/rodesai/repos","events_url":"https://api.github.com/users/rodesai/events{/privacy}","received_events_url":"https://api.github.com/users/rodesai/received_events","type":"User","site_admin":false},"repo":{"id":124003752,"node_id":"MDEwOlJlcG9zaXRvcnkxMjQwMDM3NTI=","name":"ksql","full_name":"rodesai/ksql","owner":{"login":"rodesai","id":2610028,"node_id":"MDQ6VXNlcjI2MTAwMjg=","avatar_url":"https://avatars0.githubusercontent.com/u/2610028?v=4","gravatar_id":"","url":"https://api.github.com/users/rodesai","html_url":"https://github.com/rodesai","followers_url":"https://api.github.com/users/rodesai/followers","following_url":"https://api.github.com/users/rodesai/following{/other_user}","gists_url":"https://api.github.com/users/rodesai/gists{/gist_id}","starred_url":"https://api.github.com/users/rodesai/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/rodesai/subscriptions","organizations_url":"https://api.github.com/users/rodesai/orgs","repos_url":"https://api.github.com/users/rodesai/repos","events_url":"https://api.github.com/users/rodesai/events{/privacy}","received_events_url":"https://api.github.com/users/rodesai/received_events","type":"User","site_admin":false},"private":false,"html_url":"https://github.com/rodesai/ksql","description":"KSQL - the Streaming SQL Engine for Apache Kafka ","fork":true,"url":"https://api.github.com/repos/rodesai/ksql","forks_url":"https://api.github.com/repos/rodesai/ksql/forks","keys_url":"https://api.github.com/repos/rodesai/ksql/keys{/key_id}","collaborators_url":"https://api.github.com/repos/rodesai/ksql/collaborators{/collaborator}","teams_url":"https://api.github.com/repos/rodesai/ksql/teams","hooks_url":"https://api.github.com/repos/rodesai/ksql/hooks","issue_events_url":"https://api.github.com/repos/rodesai/ksql/issues/events{/number}","events_url":"https://api.github.com/repos/rodesai/ksql/events","assignees_url":"https://api.github.com/repos/rodesai/ksql/assignees{/user}","branches_url":"https://api.github.com/repos/rodesai/ksql/branches{/branch}","tags_url":"https://api.github.com/repos/rodesai/ksql/tags","blobs_url":"https://api.github.com/repos/rodesai/ksql/git/blobs{/sha}","git_tags_url":"https://api.github.com/repos/rodesai/ksql/git/tags{/sha}","git_refs_url":"https://api.github.com/repos/rodesai/ksql/git/refs{/sha}","trees_url":"https://api.github.com/repos/rodesai/ksql/git/trees{/sha}","statuses_url":"https://api.github.com/repos/rodesai/ksql/statuses/{sha}","languages_url":"https://api.github.com/repos/rodesai/ksql/languages","stargazers_url":"https://api.github.com/repos/rodesai/ksql/stargazers","contributors_url":"https://api.github.com/repos/rodesai/ksql/contributors","subscribers_url":"https://api.github.com/repos/rodesai/ksql/subscribers","subscription_url":"https://api.github.com/repos/rodesai/ksql/subscription","commits_url":"https://api.github.com/repos/rodesai/ksql/commits{/sha}","git_commits_url":"https://api.github.com/repos/rodesai/ksql/git/commits{/sha}","comments_url":"https://api.github.com/repos/rodesai/ksql/comments{/number}","issue_comment_url":"https://api.github.com/repos/rodesai/ksql/issues/comments{/number}","contents_url":"https://api.github.com/repos/rodesai/ksql/contents/{+path}","compare_url":"https://api.github.com/repos/rodesai/ksql/compare/{base}...{head}","merges_url":"https://api.github.com/repos/rodesai/ksql/merges","archive_url":"https://api.github.com/repos/rodesai/ksql/{archive_format}{/ref}","downloads_url":"https://api.github.com/repos/rodesai/ksql/downloads","issues_url":"https://api.github.com/repos/rodesai/ksql/issues{/number}","pulls_url":"https://api.github.com/repos/rodesai/ksql/pulls{/number}","milestones_url":"https://api.github.com/repos/rodesai/ksql/milestones{/number}","notifications_url":"https://api.github.com/repos/rodesai/ksql/notifications{?since,all,participating}","labels_url":"https://api.github.com/repos/rodesai/ksql/labels{/name}","releases_url":"https://api.github.com/repos/rodesai/ksql/releases{/id}","deployments_url":"https://api.github.com/repos/rodesai/ksql/deployments","created_at":"2018-03-06T01:44:11Z","updated_at":"2018-09-10T16:34:33Z","pushed_at":"2018-09-10T16:34:29Z","git_url":"git://github.com/rodesai/ksql.git","ssh_url":"git@github.com:rodesai/ksql.git","clone_url":"https://github.com/rodesai/ksql.git","svn_url":"https://github.com/rodesai/ksql","homepage":"","size":10780,"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","node_id":"MDc6TGljZW5zZTI="},"forks":0,"open_issues":0,"watchers":0,"default_branch":"master"}},"base":{"label":"confluentinc:master","ref":"master","sha":"24a4070751506423ea322cc98af981e42891363c","user":{"login":"confluentinc","id":9439498,"node_id":"MDEyOk9yZ2FuaXphdGlvbjk0Mzk0OTg=","avatar_url":"https://avatars0.githubusercontent.com/u/9439498?v=4","gravatar_id":"","url":"https://api.github.com/users/confluentinc","html_url":"https://github.com/confluentinc","followers_url":"https://api.github.com/users/confluentinc/followers","following_url":"https://api.github.com/users/confluentinc/following{/other_user}","gists_url":"https://api.github.com/users/confluentinc/gists{/gist_id}","starred_url":"https://api.github.com/users/confluentinc/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/confluentinc/subscriptions","organizations_url":"https://api.github.com/users/confluentinc/orgs","repos_url":"https://api.github.com/users/confluentinc/repos","events_url":"https://api.github.com/users/confluentinc/events{/privacy}","received_events_url":"https://api.github.com/users/confluentinc/received_events","type":"Organization","site_admin":false},"repo":{"id":69400326,"node_id":"MDEwOlJlcG9zaXRvcnk2OTQwMDMyNg==","name":"ksql","full_name":"confluentinc/ksql","owner":{"login":"confluentinc","id":9439498,"node_id":"MDEyOk9yZ2FuaXphdGlvbjk0Mzk0OTg=","avatar_url":"https://avatars0.githubusercontent.com/u/9439498?v=4","gravatar_id":"","url":"https://api.github.com/users/confluentinc","html_url":"https://github.com/confluentinc","followers_url":"https://api.github.com/users/confluentinc/followers","following_url":"https://api.github.com/users/confluentinc/following{/other_user}","gists_url":"https://api.github.com/users/confluentinc/gists{/gist_id}","starred_url":"https://api.github.com/users/confluentinc/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/confluentinc/subscriptions","organizations_url":"https://api.github.com/users/confluentinc/orgs","repos_url":"https://api.github.com/users/confluentinc/repos","events_url":"https://api.github.com/users/confluentinc/events{/privacy}","received_events_url":"https://api.github.com/users/confluentinc/received_events","type":"Organization","site_admin":false},"private":false,"html_url":"https://github.com/confluentinc/ksql","description":"KSQL - the Streaming SQL Engine for Apache Kafka ","fork":false,"url":"https://api.github.com/repos/confluentinc/ksql","forks_url":"https://api.github.com/repos/confluentinc/ksql/forks","keys_url":"https://api.github.com/repos/confluentinc/ksql/keys{/key_id}","collaborators_url":"https://api.github.com/repos/confluentinc/ksql/collaborators{/collaborator}","teams_url":"https://api.github.com/repos/confluentinc/ksql/teams","hooks_url":"https://api.github.com/repos/confluentinc/ksql/hooks","issue_events_url":"https://api.github.com/repos/confluentinc/ksql/issues/events{/number}","events_url":"https://api.github.com/repos/confluentinc/ksql/events","assignees_url":"https://api.github.com/repos/confluentinc/ksql/assignees{/user}","branches_url":"https://api.github.com/repos/confluentinc/ksql/branches{/branch}","tags_url":"https://api.github.com/repos/confluentinc/ksql/tags","blobs_url":"https://api.github.com/repos/confluentinc/ksql/git/blobs{/sha}","git_tags_url":"https://api.github.com/repos/confluentinc/ksql/git/tags{/sha}","git_refs_url":"https://api.github.com/repos/confluentinc/ksql/git/refs{/sha}","trees_url":"https://api.github.com/repos/confluentinc/ksql/git/trees{/sha}","statuses_url":"https://api.github.com/repos/confluentinc/ksql/statuses/{sha}","languages_url":"https://api.github.com/repos/confluentinc/ksql/languages","stargazers_url":"https://api.github.com/repos/confluentinc/ksql/stargazers","contributors_url":"https://api.github.com/repos/confluentinc/ksql/contributors","subscribers_url":"https://api.github.com/repos/confluentinc/ksql/subscribers","subscription_url":"https://api.github.com/repos/confluentinc/ksql/subscription","commits_url":"https://api.github.com/repos/confluentinc/ksql/commits{/sha}","git_commits_url":"https://api.github.com/repos/confluentinc/ksql/git/commits{/sha}","comments_url":"https://api.github.com/repos/confluentinc/ksql/comments{/number}","issue_comment_url":"https://api.github.com/repos/confluentinc/ksql/issues/comments{/number}","contents_url":"https://api.github.com/repos/confluentinc/ksql/contents/{+path}","compare_url":"https://api.github.com/repos/confluentinc/ksql/compare/{base}...{head}","merges_url":"https://api.github.com/repos/confluentinc/ksql/merges","archive_url":"https://api.github.com/repos/confluentinc/ksql/{archive_format}{/ref}","downloads_url":"https://api.github.com/repos/confluentinc/ksql/downloads","issues_url":"https://api.github.com/repos/confluentinc/ksql/issues{/number}","pulls_url":"https://api.github.com/repos/confluentinc/ksql/pulls{/number}","milestones_url":"https://api.github.com/repos/confluentinc/ksql/milestones{/number}","notifications_url":"https://api.github.com/repos/confluentinc/ksql/notifications{?since,all,participating}","labels_url":"https://api.github.com/repos/confluentinc/ksql/labels{/name}","releases_url":"https://api.github.com/repos/confluentinc/ksql/releases{/id}","deployments_url":"https://api.github.com/repos/confluentinc/ksql/deployments","created_at":"2016-09-27T21:30:15Z","updated_at":"2018-09-10T13:45:08Z","pushed_at":"2018-09-10T18:06:18Z","git_url":"git://github.com/confluentinc/ksql.git","ssh_url":"git@github.com:confluentinc/ksql.git","clone_url":"https://github.com/confluentinc/ksql.git","svn_url":"https://github.com/confluentinc/ksql","homepage":"","size":10656,"stargazers_count":1895,"watchers_count":1895,"language":"Java","has_issues":true,"has_projects":true,"has_downloads":true,"has_wiki":false,"has_pages":false,"forks_count":358,"mirror_url":null,"archived":false,"open_issues_count":376,"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":358,"open_issues":376,"watchers":1895,"default_branch":"master"}},"_links":{"self":{"href":"https://api.github.com/repos/confluentinc/ksql/pulls/1856"},"html":{"href":"https://github.com/confluentinc/ksql/pull/1856"},"issue":{"href":"https://api.github.com/repos/confluentinc/ksql/issues/1856"},"comments":{"href":"https://api.github.com/repos/confluentinc/ksql/issues/1856/comments"},"review_comments":{"href":"https://api.github.com/repos/confluentinc/ksql/pulls/1856/comments"},"review_comment":{"href":"https://api.github.com/repos/confluentinc/ksql/pulls/comments{/number}"},"commits":{"href":"https://api.github.com/repos/confluentinc/ksql/pulls/1856/commits"},"statuses":{"href":"https://api.github.com/repos/confluentinc/ksql/statuses/a9cbdc49afce2601b71687d2717f19e8ecab2a1c"}},"author_association":"CONTRIBUTOR"}}
{"actor":{"display_login":"rodesai"}}