{"action":"created","comment":{"url":"https://api.github.com/repos/disqus/pgshovel/pulls/comments/38593463","id":38593463,"diff_hunk":"@@ -40,12 +54,44 @@ def consume(self, state):\n             consumer.seek(delta, 1)\n             assert consumer.offsets[0] == offset\n \n-        for offset, message in consumer:\n-            decoded = self.codec.decode(message.value)\n-            state = validate_state(state, offset, decoded)\n+        for offset, message in self.__primed(consumer, state):\n+            state = validate_state(state, offset, message)\n             # XXX: This is necessary because of a bug in protocol buffer oneof.\n             state = type(state).FromString(state.SerializeToString())\n-            yield state, offset, decoded\n+            yield state, offset, message\n+\n+    def __primed(self, consumer, state, attempts=5):\n+        decoded = imap(\n+            lambda (offset, msg): tuple((offset, self.codec.decode(msg.value))),\n+            consumer\n+        )\n+\n+        offset, message = next(decoded)\n+\n+        try:\n+            validate_state(state, offset, message)\n+        except InvalidEventError as e:\n+            # See if we can drop the first few messages in an attempt to find\n+            # the type we're looking for\n+            while attempts > 0:","path":"src/main/python/pgshovel/replication/streams/kafka.py","position":78,"original_position":78,"commit_id":"437dd4b3891ae9e3aee284f587251cb510da509a","original_commit_id":"437dd4b3891ae9e3aee284f587251cb510da509a","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":"When a `target` finishes bootstrapping its data with the `loader`, it moves over to the `stream` to start consuming mutations.  Internal to the `KafkaStream` it starts consuming from its kafka `consumer`, validating each message against the `state`, namely the `State.stream_state` state.  When the `stream` *first* starts up, there is no `stream_state`, and as such the transaction validator state machine [validates that the first operation is a `BeginOperation`](https://github.com/disqus/pgshovel/blob/replication/src/main/python/pgshovel/replication/validation/transactions.py#L141-L143).\r\n\r\nUnfortunately, there is no guarantee that the first message is a `BeginOperation`. Most likely[1] this is due to the Kafka topic containing the messages being pruned of old messages, and now the head of the log falls in the middle of a transaction.  And if the first message isn't `BeginOperation`, prior to this diff validation would fail and this code would crash, despite the fact that if the stream just seeked a few more messages, it would have found a `BeginOperation` and could have continued correctly.\r\n\r\n[1] It's also possible to call `KafkaStream.consume()` with a `state.stream_state.transaction_state` value that is incompatible with the next message from the `state`'s offset in the mutations stream - i.e. `transaction_state` is `InTransaction` and a `BeginOperation` is the first message seen - though *theoretically* this shouldn't be possible because a replication target instances should be setting `state` offset at the same time as the `transaction_state`. If this does happen, the code will see a `InvalidEventError` on line 72 and either give up trying to find a `BeginOperation` or find one, yield it out, and then will immediately fail the `valid_state` call since the transaction state machine isn't valid and the sequence number of the next message (the found `BeginOperation` isn't 1 more than the `state`'s sequence number).\r\n\r\nThis may be confusing to you so let me know if this makes sense.","created_at":"2015-09-02T22:32:20Z","updated_at":"2015-09-02T22:32:20Z","html_url":"https://github.com/disqus/pgshovel/pull/26#discussion_r38593463","pull_request_url":"https://api.github.com/repos/disqus/pgshovel/pulls/26","_links":{"self":{"href":"https://api.github.com/repos/disqus/pgshovel/pulls/comments/38593463"},"html":{"href":"https://github.com/disqus/pgshovel/pull/26#discussion_r38593463"},"pull_request":{"href":"https://api.github.com/repos/disqus/pgshovel/pulls/26"}}},"pull_request":{"url":"https://api.github.com/repos/disqus/pgshovel/pulls/26","id":43818956,"html_url":"https://github.com/disqus/pgshovel/pull/26","diff_url":"https://github.com/disqus/pgshovel/pull/26.diff","patch_url":"https://github.com/disqus/pgshovel/pull/26.patch","issue_url":"https://api.github.com/repos/disqus/pgshovel/issues/26","number":26,"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":"Fixes #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.","created_at":"2015-08-31T22:39:55Z","updated_at":"2015-09-02T22:32:20Z","closed_at":null,"merged_at":null,"merge_commit_sha":"0b5ae7ca404f44cd8b0e0a3facebfc8daae2b00d","assignee":null,"milestone":{"url":"https://api.github.com/repos/disqus/pgshovel/milestones/2","html_url":"https://github.com/disqus/pgshovel/milestones/0.9","labels_url":"https://api.github.com/repos/disqus/pgshovel/milestones/2/labels","id":1277982,"number":2,"title":"0.9","description":"Represents a version containing replication tooling and no major known bugs, but with some known issues (mostly administrative and/or operational in nature) intended to be used for non-mission critical use cases where hours (or days) of downtime and potential data corruption (possibly irrecoverable) due to operator error can occur.","creator":{"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},"open_issues":5,"closed_issues":2,"state":"open","created_at":"2015-08-28T17:20:08Z","updated_at":"2015-09-02T16:03:06Z","due_on":null,"closed_at":null},"commits_url":"https://api.github.com/repos/disqus/pgshovel/pulls/26/commits","review_comments_url":"https://api.github.com/repos/disqus/pgshovel/pulls/26/comments","review_comment_url":"https://api.github.com/repos/disqus/pgshovel/pulls/comments{/number}","comments_url":"https://api.github.com/repos/disqus/pgshovel/issues/26/comments","statuses_url":"https://api.github.com/repos/disqus/pgshovel/statuses/437dd4b3891ae9e3aee284f587251cb510da509a","head":{"label":"disqus:replication_agnostic_stream_start","ref":"replication_agnostic_stream_start","sha":"437dd4b3891ae9e3aee284f587251cb510da509a","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-02T00:31:56Z","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":676,"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":24,"forks":1,"open_issues":24,"watchers":2,"default_branch":"master"}},"base":{"label":"disqus:new_tests","ref":"new_tests","sha":"d73035224dc0b062347bff8d55604e518030b373","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-02T00:31:56Z","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":676,"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":24,"forks":1,"open_issues":24,"watchers":2,"default_branch":"master"}},"_links":{"self":{"href":"https://api.github.com/repos/disqus/pgshovel/pulls/26"},"html":{"href":"https://github.com/disqus/pgshovel/pull/26"},"issue":{"href":"https://api.github.com/repos/disqus/pgshovel/issues/26"},"comments":{"href":"https://api.github.com/repos/disqus/pgshovel/issues/26/comments"},"review_comments":{"href":"https://api.github.com/repos/disqus/pgshovel/pulls/26/comments"},"review_comment":{"href":"https://api.github.com/repos/disqus/pgshovel/pulls/comments{/number}"},"commits":{"href":"https://api.github.com/repos/disqus/pgshovel/pulls/26/commits"},"statuses":{"href":"https://api.github.com/repos/disqus/pgshovel/statuses/437dd4b3891ae9e3aee284f587251cb510da509a"}}}}