fix: Fix the resuming of MongoDB change streams in log-based replication mode #31

menzenski · 2024-02-16T00:46:36Z

Fixes #27
Fixes #29
Fixes #28
Fixes #30

…ally

…-mongodb into change-stream-improvements

pyproject.toml

menzenski · 2024-02-16T00:53:33Z

tap_mongodb/streams.py

@@ -179,7 +170,7 @@ def _generate_record_messages(self, record: dict) -> Generator[singer.RecordMess
            Record message objects.
        """
        extracted_at: DateTime = record.pop("_sdc_extracted_at", utc_now())
-        pop_deselected_record_properties(record, self.schema, self.mask, self.logger)
+        pop_deselected_record_properties(record, self.schema, self.mask)


Updating the SDK version requires this change.

Sorry about this change, I was confident this was in practice a private API 😞

menzenski · 2024-02-16T00:54:04Z

tap_mongodb/streams.py

@@ -197,174 +188,190 @@ def _generate_record_messages(self, record: dict) -> Generator[singer.RecordMess
                )
                yield record_message

-    def get_records(self, context: dict | None) -> Iterable[dict]:
-        """Return a generator of record-type dictionary objects."""
+    def _get_records_incremental(


refactored slightly to split up the giant method, now have incremental and log-based get_records implementation methods.

menzenski · 2024-02-16T00:58:43Z

tap_mongodb/streams.py

+            self.logger.critical(exception)
+            raise exception
+
+        while change_stream.alive and keep_open:


it's hard to see in this diff with the other refactors, but note the with change_stream in https://github.com/MeltanoLabs/tap-mongodb/pull/31/files#diff-e276f9374d5b67980ff79f28e37aa6417f4f456c8bdfbf49d709acd30ebb3f7aL289 that is absent here.

menzenski · 2024-02-16T01:04:38Z

tap_mongodb/streams.py

-                        ) from operation_failure
-                elif (
-                    self._connector.version < (4, 2)
+                    resume_strategy == ResumeStrategy.RESUME_AFTER
                    and operation_failure.code == 286
                    and "as the resume point may no longer be in the oplog." in operation_failure.details["errmsg"]
                ):
                    self.logger.warning("Unable to resume change stream from resume token. Resetting resume token.")
                    change_stream_options.pop("resume_after", None)


The diff with the refactor makes this hard to see, but lines 291–293 here are new - these three lines combined with the removal of the with change_stream line (see above comment) are the meat of this fix. These lines allow the tap to gracefully re-open a change stream if the resume token from the state is not valid.

.github/workflows/validate.yml

pyproject.toml

menzenski · 2024-02-16T01:11:13Z

tap_mongodb/streams.py

+                # fullDocument key is not present on delete events - if it is missing, fall back to documentKey
+                # instead. If that is missing, pass None/null to avoid raising an error.
+                document = record.get("fullDocument", record.get("documentKey", None))
+                object_id: Optional[ObjectId] = document["_id"] if document and "_id" in document else None


#27

this issue is fixed with the if document check. Compare:

object_id: Optional[ObjectId] = document["_id"] if "_id" in document else None

to

object_id: Optional[ObjectId] = document["_id"] if document and "_id" in document else None

pyproject.toml

menzenski · 2024-02-17T14:26:12Z

pyproject.toml

@@ -28,7 +29,7 @@ black = "^23.1.0"
 pyupgrade = "^3.3.1"
 mypy = "^1.0.0"
 isort = "^5.11.5"
-singer-sdk = { version = "^0.31.1", extras = ["testing"] }
+singer-sdk = { version = "^0.35.0", extras = ["testing"] }


edgarrmondragon · 2024-02-19T22:16:45Z

tap_mongodb/connector.py

@@ -43,7 +35,7 @@ def __init__(  # pylint: disable=too-many-arguments
        self._datetime_conversion: str = datetime_conversion.upper()
        self._prefix: Optional[str] = prefix
        self._logger: Logger = getLogger(__name__)
-        self._version: Optional[MongoVersion] = None
+        self._version: MongoVersion


Might make sense to move this annotation to the class level, ie

class MongoDBConnector: _version: MongoVersion

edgarrmondragon · 2024-02-19T22:33:45Z

tap_mongodb/streams.py

@@ -179,7 +170,7 @@ def _generate_record_messages(self, record: dict) -> Generator[singer.RecordMess
            Record message objects.
        """
        extracted_at: DateTime = record.pop("_sdc_extracted_at", utc_now())
-        pop_deselected_record_properties(record, self.schema, self.mask, self.logger)
+        pop_deselected_record_properties(record, self.schema, self.mask)


Sorry about this change, I was confident this was in practice a private API 😞

edgarrmondragon · 2024-02-19T22:54:26Z

tap_mongodb/connector.py


 from pymongo import MongoClient
 from pymongo.database import Database
 from pymongo.errors import PyMongoError
 from singer_sdk._singerlib.catalog import CatalogEntry, MetadataMapping, Schema

 from tap_mongodb.schema import SCHEMA
+from tap_mongodb.types import MongoVersion

 if sys.version_info[:2] < (3, 8):
    from backports.cached_property import cached_property
 else:
    from functools import cached_property


Since we're dropping support for Python 3.7 I think this can be simplified

edgarrmondragon · 2024-02-19T22:56:50Z

tap_mongodb/streams.py

@@ -102,7 +96,6 @@ def primary_keys(self) -> Optional[list[str]]:
    def primary_keys(self, new_value: list[str]) -> None:
        """Set primary keys for the stream."""
        self._primary_keys = new_value
-        return self


tap_mongodb/streams.py

Co-authored-by: Edgar Ramírez Mondragón <[email protected]>

menzenski · 2024-02-20T11:47:27Z

Moving this PR to Draft status - I'm still working on it.

…-mongodb into change-stream-improvements

edgarrmondragon · 2024-02-23T23:56:13Z

Moving this PR to Draft status - I'm still working on it.

👍 @menzenski thanks for this PR! Ping me when it's ready :)

menzenski · 2024-02-23T23:58:06Z

@edgarrmondragon will do!

I think I have a way to run a MongoDB container in docker as a replica set, which hopefully will let me add real actual tests of the log-based behavior.

…mmy record scenario

menzenski added 15 commits November 17, 2023 09:34

Start scaffolding things to test change stream handling more specific…

a0280ce

…ally

some more refactoring, start adding resume strategy setting and tests

6ea1f0c

more tests, start splitting up the giant method

8d61b87

add mypy to precommit checks

4f0b291

add debugging logging config file

48e9610

update sdk version

d3aacb9

did this fix it?

154b62e

delete logging config

fa6ad4c

delete mongodb testing things, they didn't work with change streams

32f6a2a

remove start at operation time

2f9c015

rm seed utils

650ec30

bump version

c4335e5

remove backports cached property if we're on 3.8+ now

b0e253a

Fix change stream handling in MongoDB log-based replication mode

cf875b8

Merge branch 'change-stream-improvements' of github.com:menzenski/tap…

aaf7dd7

…-mongodb into change-stream-improvements

menzenski self-assigned this Feb 16, 2024

menzenski added 2 commits February 15, 2024 18:51

remove loguru, not used

2a3c032

rm seed command

8fac23a

menzenski commented Feb 16, 2024

View reviewed changes

test with python 3.12

96d6811

menzenski commented Feb 16, 2024

View reviewed changes

menzenski added 4 commits February 15, 2024 19:12

more python version updates

83fa85e

rm generic reference to re.match

f2ac545

rm conftest

b1f293b

more strategy cleanup

0648fff

menzenski changed the title ~~Fix the resuming of MongoDB change streams in log-based replication mode~~ fix: Fix the resuming of MongoDB change streams in log-based replication mode Feb 16, 2024

menzenski requested review from WillDaSilva, edgarrmondragon and a team February 16, 2024 16:11

menzenski commented Feb 17, 2024

View reviewed changes

add remaining change event fields

2451443

edgarrmondragon reviewed Feb 19, 2024

View reviewed changes

Update streams.py

316daa4

Co-authored-by: Edgar Ramírez Mondragón <[email protected]>

menzenski marked this pull request as draft February 20, 2024 11:46

menzenski added 2 commits February 20, 2024 06:30

add a whole bunch of mongo-replica-set-in-docker experiments

794c2da

let's see if that gets a mongodb replica set stood up in github actions

d7a6250

menzenski marked this pull request as ready for review February 20, 2024 13:12

Merge branch 'change-stream-improvements' of github.com:menzenski/tap…

a030376

…-mongodb into change-stream-improvements

menzenski marked this pull request as draft February 20, 2024 13:12

menzenski added 3 commits February 20, 2024 07:14

quote version numbers

9ec4ba4

support 3.12 patch versions

9f3d774

don't always throw an exception

a1b8257

menzenski added 13 commits April 23, 2024 06:05

re-lock plugins

43272cd

re-lock poetry dependencies, update local python version

4ff5b89

first pass at tests that use testcontainers-mongodb

4310b48

Lint and docstrings

a85f1dd

add a real test case of the tap in incremental mode

981523d

rm that block

1952c29

make integration_tests a package so we can import from it

b161e20

try getting replica set to work

f481319

rm replica-set attempts, didn't work

25f69ba

use loguru for all logging because I understand it

d42a173

various lint and typing fixes

0207c76

add collection name to log messages and more explicitly handle the du…

95a88b3

…mmy record scenario

set metadata on a dummy record

909078d

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

fix: Fix the resuming of MongoDB change streams in log-based replication mode #31

fix: Fix the resuming of MongoDB change streams in log-based replication mode #31

menzenski commented Feb 16, 2024 •

edited

Loading

menzenski Feb 16, 2024

edgarrmondragon Feb 19, 2024

menzenski Feb 16, 2024

menzenski Feb 16, 2024

menzenski Feb 16, 2024

menzenski Feb 16, 2024

menzenski Feb 17, 2024

edgarrmondragon Feb 19, 2024

edgarrmondragon Feb 19, 2024

edgarrmondragon Feb 19, 2024

edgarrmondragon Feb 19, 2024

menzenski commented Feb 20, 2024

edgarrmondragon commented Feb 23, 2024

menzenski commented Feb 23, 2024

fix: Fix the resuming of MongoDB change streams in log-based replication mode #31

Are you sure you want to change the base?

fix: Fix the resuming of MongoDB change streams in log-based replication mode #31

Conversation

menzenski commented Feb 16, 2024 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

menzenski commented Feb 20, 2024

edgarrmondragon commented Feb 23, 2024

menzenski commented Feb 23, 2024

menzenski commented Feb 16, 2024 •

edited

Loading