{"action":"created","comment":{"url":"https://api.github.com/repos/disqus/pgshovel/pulls/comments/39101928","id":39101928,"diff_hunk":"@@ -29,7 +36,12 @@ def consume(self, state):\n         # hard we don't really need to deal with it right now.\n         assert len(consumer.offsets) is 1\n \n-        if state.HasField('stream_state'):\n+        decoded = imap(\n+            lambda (offset, msg): tuple((offset, self.codec.decode(msg.value))),","path":"src/main/python/pgshovel/replication/streams/kafka.py","position":41,"original_position":41,"commit_id":"468c67bc5703aac9f768abc3cfef625b479a1eb5","original_commit_id":"468c67bc5703aac9f768abc3cfef625b479a1eb5","user":{"login":"tkaemming","id":65315,"avatar_url":"https://avatars.githubusercontent.com/u/65315?v=3","gravatar_id":"","url":"https://api.github.com/users/tkaemming","html_url":"https://github.com/tkaemming","followers_url":"https://api.github.com/users/tkaemming/followers","following_url":"https://api.github.com/users/tkaemming/following{/other_user}","gists_url":"https://api.github.com/users/tkaemming/gists{/gist_id}","starred_url":"https://api.github.com/users/tkaemming/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/tkaemming/subscriptions","organizations_url":"https://api.github.com/users/tkaemming/orgs","repos_url":"https://api.github.com/users/tkaemming/repos","events_url":"https://api.github.com/users/tkaemming/events{/privacy}","received_events_url":"https://api.github.com/users/tkaemming/received_events","type":"User","site_admin":false},"body":"`tuple` constructor is redundant here:\r\n\r\n```\r\n>>> tuple((1,2)) == (1,2)\r\nTrue\r\n```","created_at":"2015-09-09T21:55:40Z","updated_at":"2015-09-09T21:55:40Z","html_url":"https://github.com/disqus/pgshovel/pull/30#discussion_r39101928","pull_request_url":"https://api.github.com/repos/disqus/pgshovel/pulls/30","_links":{"self":{"href":"https://api.github.com/repos/disqus/pgshovel/pulls/comments/39101928"},"html":{"href":"https://github.com/disqus/pgshovel/pull/30#discussion_r39101928"},"pull_request":{"href":"https://api.github.com/repos/disqus/pgshovel/pulls/30"}}},"pull_request":{"url":"https://api.github.com/repos/disqus/pgshovel/pulls/30","id":44592631,"html_url":"https://github.com/disqus/pgshovel/pull/30","diff_url":"https://github.com/disqus/pgshovel/pull/30.diff","patch_url":"https://github.com/disqus/pgshovel/pull/30.patch","issue_url":"https://api.github.com/repos/disqus/pgshovel/issues/30","number":30,"state":"open","locked":false,"title":"Replication agnostic stream start","user":{"login":"Fluxx","id":5892,"avatar_url":"https://avatars.githubusercontent.com/u/5892?v=3","gravatar_id":"","url":"https://api.github.com/users/Fluxx","html_url":"https://github.com/Fluxx","followers_url":"https://api.github.com/users/Fluxx/followers","following_url":"https://api.github.com/users/Fluxx/following{/other_user}","gists_url":"https://api.github.com/users/Fluxx/gists{/gist_id}","starred_url":"https://api.github.com/users/Fluxx/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/Fluxx/subscriptions","organizations_url":"https://api.github.com/users/Fluxx/orgs","repos_url":"https://api.github.com/users/Fluxx/repos","events_url":"https://api.github.com/users/Fluxx/events{/privacy}","received_events_url":"https://api.github.com/users/Fluxx/received_events","type":"User","site_admin":false},"body":"Continuation of the old (now closed) #26 pull request.  The reason this one is new is that the upstream branch that #26 was based off of, `new_tests` was merged into the `replication` branch, but it is impossible to change upstream branches of a pull request once it's created. Additionally, I tried to rebase the PR's branch off of `replication` branch and force push it, and was unable to re-open the pull request since I force pushed and/or recreated the branch (according to the Github UI).\r\n\r\nIn light of all of this this decided the cleanest way was just to reopen the branch with a new pull request.\r\n\r\n---\r\n\r\nFixes #10 by doing two things:\r\n\r\n1. Adds an `expected` property to the `InvalidEventError`, which contains the expected valid event given the state the transaction state machine is in.\r\n2. Adds code to the `replication.streams.kafka:KafkaStream` class to \"prime\" the consumer stream.  Priming:\r\n  1. Attempts to validate the first message in the stream.\r\n  2. If the message is valid, it is yielded out.\r\n  3. If it is not valid, and the exception contains `expected` events, the `KafkaStream` peeks at the first few messages in the stream looking for one matching an `expected` type.\r\n  4. If a message is found, that message is yielded and iteration of the stream picks up from there.\r\n  5. If a message is not found the original exception is raised.\r\n\r\nPlease see #10 for more detail on the error.\r\n\r\nAdditionally, when I merge this code into the `replication` branch I will make sure it's a fast forward.","created_at":"2015-09-09T19:54:21Z","updated_at":"2015-09-09T21:55:40Z","closed_at":null,"merged_at":null,"merge_commit_sha":"05f49a1614293464d462752893eb7c6e5ddb6b15","assignee":null,"milestone":null,"commits_url":"https://api.github.com/repos/disqus/pgshovel/pulls/30/commits","review_comments_url":"https://api.github.com/repos/disqus/pgshovel/pulls/30/comments","review_comment_url":"https://api.github.com/repos/disqus/pgshovel/pulls/comments{/number}","comments_url":"https://api.github.com/repos/disqus/pgshovel/issues/30/comments","statuses_url":"https://api.github.com/repos/disqus/pgshovel/statuses/468c67bc5703aac9f768abc3cfef625b479a1eb5","head":{"label":"disqus:replication_agnostic_stream_start","ref":"replication_agnostic_stream_start","sha":"468c67bc5703aac9f768abc3cfef625b479a1eb5","user":{"login":"disqus","id":317745,"avatar_url":"https://avatars.githubusercontent.com/u/317745?v=3","gravatar_id":"","url":"https://api.github.com/users/disqus","html_url":"https://github.com/disqus","followers_url":"https://api.github.com/users/disqus/followers","following_url":"https://api.github.com/users/disqus/following{/other_user}","gists_url":"https://api.github.com/users/disqus/gists{/gist_id}","starred_url":"https://api.github.com/users/disqus/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/disqus/subscriptions","organizations_url":"https://api.github.com/users/disqus/orgs","repos_url":"https://api.github.com/users/disqus/repos","events_url":"https://api.github.com/users/disqus/events{/privacy}","received_events_url":"https://api.github.com/users/disqus/received_events","type":"Organization","site_admin":false},"repo":{"id":40566869,"name":"pgshovel","full_name":"disqus/pgshovel","owner":{"login":"disqus","id":317745,"avatar_url":"https://avatars.githubusercontent.com/u/317745?v=3","gravatar_id":"","url":"https://api.github.com/users/disqus","html_url":"https://github.com/disqus","followers_url":"https://api.github.com/users/disqus/followers","following_url":"https://api.github.com/users/disqus/following{/other_user}","gists_url":"https://api.github.com/users/disqus/gists{/gist_id}","starred_url":"https://api.github.com/users/disqus/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/disqus/subscriptions","organizations_url":"https://api.github.com/users/disqus/orgs","repos_url":"https://api.github.com/users/disqus/repos","events_url":"https://api.github.com/users/disqus/events{/privacy}","received_events_url":"https://api.github.com/users/disqus/received_events","type":"Organization","site_admin":false},"private":false,"html_url":"https://github.com/disqus/pgshovel","description":"A change data capture system for PostgreSQL","fork":false,"url":"https://api.github.com/repos/disqus/pgshovel","forks_url":"https://api.github.com/repos/disqus/pgshovel/forks","keys_url":"https://api.github.com/repos/disqus/pgshovel/keys{/key_id}","collaborators_url":"https://api.github.com/repos/disqus/pgshovel/collaborators{/collaborator}","teams_url":"https://api.github.com/repos/disqus/pgshovel/teams","hooks_url":"https://api.github.com/repos/disqus/pgshovel/hooks","issue_events_url":"https://api.github.com/repos/disqus/pgshovel/issues/events{/number}","events_url":"https://api.github.com/repos/disqus/pgshovel/events","assignees_url":"https://api.github.com/repos/disqus/pgshovel/assignees{/user}","branches_url":"https://api.github.com/repos/disqus/pgshovel/branches{/branch}","tags_url":"https://api.github.com/repos/disqus/pgshovel/tags","blobs_url":"https://api.github.com/repos/disqus/pgshovel/git/blobs{/sha}","git_tags_url":"https://api.github.com/repos/disqus/pgshovel/git/tags{/sha}","git_refs_url":"https://api.github.com/repos/disqus/pgshovel/git/refs{/sha}","trees_url":"https://api.github.com/repos/disqus/pgshovel/git/trees{/sha}","statuses_url":"https://api.github.com/repos/disqus/pgshovel/statuses/{sha}","languages_url":"https://api.github.com/repos/disqus/pgshovel/languages","stargazers_url":"https://api.github.com/repos/disqus/pgshovel/stargazers","contributors_url":"https://api.github.com/repos/disqus/pgshovel/contributors","subscribers_url":"https://api.github.com/repos/disqus/pgshovel/subscribers","subscription_url":"https://api.github.com/repos/disqus/pgshovel/subscription","commits_url":"https://api.github.com/repos/disqus/pgshovel/commits{/sha}","git_commits_url":"https://api.github.com/repos/disqus/pgshovel/git/commits{/sha}","comments_url":"https://api.github.com/repos/disqus/pgshovel/comments{/number}","issue_comment_url":"https://api.github.com/repos/disqus/pgshovel/issues/comments{/number}","contents_url":"https://api.github.com/repos/disqus/pgshovel/contents/{+path}","compare_url":"https://api.github.com/repos/disqus/pgshovel/compare/{base}...{head}","merges_url":"https://api.github.com/repos/disqus/pgshovel/merges","archive_url":"https://api.github.com/repos/disqus/pgshovel/{archive_format}{/ref}","downloads_url":"https://api.github.com/repos/disqus/pgshovel/downloads","issues_url":"https://api.github.com/repos/disqus/pgshovel/issues{/number}","pulls_url":"https://api.github.com/repos/disqus/pgshovel/pulls{/number}","milestones_url":"https://api.github.com/repos/disqus/pgshovel/milestones{/number}","notifications_url":"https://api.github.com/repos/disqus/pgshovel/notifications{?since,all,participating}","labels_url":"https://api.github.com/repos/disqus/pgshovel/labels{/name}","releases_url":"https://api.github.com/repos/disqus/pgshovel/releases{/id}","created_at":"2015-08-11T21:49:27Z","updated_at":"2015-09-02T15:26:36Z","pushed_at":"2015-09-09T20:18:19Z","git_url":"git://github.com/disqus/pgshovel.git","ssh_url":"git@github.com:disqus/pgshovel.git","clone_url":"https://github.com/disqus/pgshovel.git","svn_url":"https://github.com/disqus/pgshovel","homepage":"","size":607,"stargazers_count":2,"watchers_count":2,"language":"Python","has_issues":true,"has_downloads":true,"has_wiki":true,"has_pages":false,"forks_count":1,"mirror_url":null,"open_issues_count":22,"forks":1,"open_issues":22,"watchers":2,"default_branch":"master"}},"base":{"label":"disqus:replication","ref":"replication","sha":"371d315d7e06d293402280eaf5c81e5c0c31663a","user":{"login":"disqus","id":317745,"avatar_url":"https://avatars.githubusercontent.com/u/317745?v=3","gravatar_id":"","url":"https://api.github.com/users/disqus","html_url":"https://github.com/disqus","followers_url":"https://api.github.com/users/disqus/followers","following_url":"https://api.github.com/users/disqus/following{/other_user}","gists_url":"https://api.github.com/users/disqus/gists{/gist_id}","starred_url":"https://api.github.com/users/disqus/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/disqus/subscriptions","organizations_url":"https://api.github.com/users/disqus/orgs","repos_url":"https://api.github.com/users/disqus/repos","events_url":"https://api.github.com/users/disqus/events{/privacy}","received_events_url":"https://api.github.com/users/disqus/received_events","type":"Organization","site_admin":false},"repo":{"id":40566869,"name":"pgshovel","full_name":"disqus/pgshovel","owner":{"login":"disqus","id":317745,"avatar_url":"https://avatars.githubusercontent.com/u/317745?v=3","gravatar_id":"","url":"https://api.github.com/users/disqus","html_url":"https://github.com/disqus","followers_url":"https://api.github.com/users/disqus/followers","following_url":"https://api.github.com/users/disqus/following{/other_user}","gists_url":"https://api.github.com/users/disqus/gists{/gist_id}","starred_url":"https://api.github.com/users/disqus/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/disqus/subscriptions","organizations_url":"https://api.github.com/users/disqus/orgs","repos_url":"https://api.github.com/users/disqus/repos","events_url":"https://api.github.com/users/disqus/events{/privacy}","received_events_url":"https://api.github.com/users/disqus/received_events","type":"Organization","site_admin":false},"private":false,"html_url":"https://github.com/disqus/pgshovel","description":"A change data capture system for PostgreSQL","fork":false,"url":"https://api.github.com/repos/disqus/pgshovel","forks_url":"https://api.github.com/repos/disqus/pgshovel/forks","keys_url":"https://api.github.com/repos/disqus/pgshovel/keys{/key_id}","collaborators_url":"https://api.github.com/repos/disqus/pgshovel/collaborators{/collaborator}","teams_url":"https://api.github.com/repos/disqus/pgshovel/teams","hooks_url":"https://api.github.com/repos/disqus/pgshovel/hooks","issue_events_url":"https://api.github.com/repos/disqus/pgshovel/issues/events{/number}","events_url":"https://api.github.com/repos/disqus/pgshovel/events","assignees_url":"https://api.github.com/repos/disqus/pgshovel/assignees{/user}","branches_url":"https://api.github.com/repos/disqus/pgshovel/branches{/branch}","tags_url":"https://api.github.com/repos/disqus/pgshovel/tags","blobs_url":"https://api.github.com/repos/disqus/pgshovel/git/blobs{/sha}","git_tags_url":"https://api.github.com/repos/disqus/pgshovel/git/tags{/sha}","git_refs_url":"https://api.github.com/repos/disqus/pgshovel/git/refs{/sha}","trees_url":"https://api.github.com/repos/disqus/pgshovel/git/trees{/sha}","statuses_url":"https://api.github.com/repos/disqus/pgshovel/statuses/{sha}","languages_url":"https://api.github.com/repos/disqus/pgshovel/languages","stargazers_url":"https://api.github.com/repos/disqus/pgshovel/stargazers","contributors_url":"https://api.github.com/repos/disqus/pgshovel/contributors","subscribers_url":"https://api.github.com/repos/disqus/pgshovel/subscribers","subscription_url":"https://api.github.com/repos/disqus/pgshovel/subscription","commits_url":"https://api.github.com/repos/disqus/pgshovel/commits{/sha}","git_commits_url":"https://api.github.com/repos/disqus/pgshovel/git/commits{/sha}","comments_url":"https://api.github.com/repos/disqus/pgshovel/comments{/number}","issue_comment_url":"https://api.github.com/repos/disqus/pgshovel/issues/comments{/number}","contents_url":"https://api.github.com/repos/disqus/pgshovel/contents/{+path}","compare_url":"https://api.github.com/repos/disqus/pgshovel/compare/{base}...{head}","merges_url":"https://api.github.com/repos/disqus/pgshovel/merges","archive_url":"https://api.github.com/repos/disqus/pgshovel/{archive_format}{/ref}","downloads_url":"https://api.github.com/repos/disqus/pgshovel/downloads","issues_url":"https://api.github.com/repos/disqus/pgshovel/issues{/number}","pulls_url":"https://api.github.com/repos/disqus/pgshovel/pulls{/number}","milestones_url":"https://api.github.com/repos/disqus/pgshovel/milestones{/number}","notifications_url":"https://api.github.com/repos/disqus/pgshovel/notifications{?since,all,participating}","labels_url":"https://api.github.com/repos/disqus/pgshovel/labels{/name}","releases_url":"https://api.github.com/repos/disqus/pgshovel/releases{/id}","created_at":"2015-08-11T21:49:27Z","updated_at":"2015-09-02T15:26:36Z","pushed_at":"2015-09-09T20:18:19Z","git_url":"git://github.com/disqus/pgshovel.git","ssh_url":"git@github.com:disqus/pgshovel.git","clone_url":"https://github.com/disqus/pgshovel.git","svn_url":"https://github.com/disqus/pgshovel","homepage":"","size":607,"stargazers_count":2,"watchers_count":2,"language":"Python","has_issues":true,"has_downloads":true,"has_wiki":true,"has_pages":false,"forks_count":1,"mirror_url":null,"open_issues_count":22,"forks":1,"open_issues":22,"watchers":2,"default_branch":"master"}},"_links":{"self":{"href":"https://api.github.com/repos/disqus/pgshovel/pulls/30"},"html":{"href":"https://github.com/disqus/pgshovel/pull/30"},"issue":{"href":"https://api.github.com/repos/disqus/pgshovel/issues/30"},"comments":{"href":"https://api.github.com/repos/disqus/pgshovel/issues/30/comments"},"review_comments":{"href":"https://api.github.com/repos/disqus/pgshovel/pulls/30/comments"},"review_comment":{"href":"https://api.github.com/repos/disqus/pgshovel/pulls/comments{/number}"},"commits":{"href":"https://api.github.com/repos/disqus/pgshovel/pulls/30/commits"},"statuses":{"href":"https://api.github.com/repos/disqus/pgshovel/statuses/468c67bc5703aac9f768abc3cfef625b479a1eb5"}}}}