From 90321d778e70fe2bdcaf2d383e04c7c42e3efc61 Mon Sep 17 00:00:00 2001 From: Matt Menzenski Date: Tue, 17 Oct 2023 14:57:35 -0500 Subject: [PATCH 1/3] Use start_after instead of resume_after for change stream tokens if MongoDB version >= 4.2 --- pyproject.toml | 2 +- tap_mongodb/connector.py | 15 +++++++++++++-- tap_mongodb/streams.py | 14 +++++++++++--- 3 files changed, 25 insertions(+), 6 deletions(-) diff --git a/pyproject.toml b/pyproject.toml index 6395c88..30ab036 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -1,6 +1,6 @@ [tool.poetry] name = "tap-mongodb" -version = "2.3.3" +version = "2.4.0" description = "`tap-mongodb` is a Singer tap for MongoDB and AWS DocumentDB, built with the Meltano Singer SDK." readme = "README.md" authors = ["Matt Menzenski"] diff --git a/tap_mongodb/connector.py b/tap_mongodb/connector.py index 366192a..627d141 100644 --- a/tap_mongodb/connector.py +++ b/tap_mongodb/connector.py @@ -2,7 +2,7 @@ import sys from logging import Logger, getLogger -from typing import Any, Dict, List, Optional +from typing import Any, Dict, List, Optional, Tuple, TypeAlias from pymongo import MongoClient from pymongo.database import Database @@ -17,6 +17,9 @@ from functools import cached_property +MongoVersion: TypeAlias = Tuple[int, int] + + class MongoDBConnector: """MongoDB/DocumentDB connector class""" @@ -34,6 +37,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 @cached_property def mongo_client(self) -> MongoClient: @@ -42,7 +46,9 @@ def mongo_client(self) -> MongoClient: self._connection_string, datetime_conversion=self._datetime_conversion, **self._options ) try: - client.server_info() + server_info: Dict[str, Any] = client.server_info() + version_array: List[int] = server_info["versionArray"] + self._version = (version_array[0], version_array[1]) except Exception as exception: self._logger.exception("Could not connect to MongoDB") raise RuntimeError("Could not connect to MongoDB") from exception @@ -53,6 +59,11 @@ def database(self) -> Database: """Provide a Database instance.""" return self.mongo_client[self._db_name] + @property + def version(self) -> Optional[MongoVersion]: + """Returns the MongoVersion that is being used.""" + return self._version + def get_fully_qualified_name( self, collection_name: str, diff --git a/tap_mongodb/streams.py b/tap_mongodb/streams.py index 2232920..21d0bf0 100644 --- a/tap_mongodb/streams.py +++ b/tap_mongodb/streams.py @@ -218,7 +218,13 @@ def get_records(self, context: dict | None) -> Iterable[dict]: change_stream_options = {"full_document": "updateLookup"} if bookmark is not None and bookmark != DEFAULT_START_DATE: self.logger.debug(f"using bookmark: {bookmark}") - change_stream_options["resume_after"] = {"_data": bookmark} + # if on mongo version 4.2 or above, use start_after instead of resume_after, as the former will + # gracefully open a new change stream if the resume token's event is not present in the oplog, while + # the latter will error in that scenario. + if self._connector.version >= (4, 2): + change_stream_options["start_after"] = {"_data": bookmark} + else: + change_stream_options["resume_after"] = {"_data": bookmark} operation_types_allowlist: set = set(self.config.get("operation_types")) has_seen_a_record: bool = False keep_open: bool = True @@ -245,7 +251,8 @@ def get_records(self, context: dict | None) -> Iterable[dict]: f"Unable to enable change streams on collection {collection.name}" ) from operation_failure elif ( - operation_failure.code == 286 + self._connector.version < (4, 2) + 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.") @@ -265,7 +272,8 @@ def get_records(self, context: dict | None) -> Iterable[dict]: record = change_stream.try_next() except OperationFailure as operation_failure: if ( - operation_failure.code == 286 + self._connector.version < (4, 2) + and operation_failure.code == 286 and "as the resume point may no longer be in the oplog." in operation_failure.details["errmsg"] ): From fb8498ec9ad7ec91b4bcf7d4e8e9e73a8ebba726 Mon Sep 17 00:00:00 2001 From: Matt Menzenski Date: Wed, 18 Oct 2023 07:20:35 -0500 Subject: [PATCH 2/3] TypeAlias only exists in Python 3.10+ --- tap_mongodb/connector.py | 10 ++++++++-- 1 file changed, 8 insertions(+), 2 deletions(-) diff --git a/tap_mongodb/connector.py b/tap_mongodb/connector.py index 627d141..ca6f35b 100644 --- a/tap_mongodb/connector.py +++ b/tap_mongodb/connector.py @@ -2,7 +2,7 @@ import sys from logging import Logger, getLogger -from typing import Any, Dict, List, Optional, Tuple, TypeAlias +from typing import Any, Dict, List, Optional, Tuple from pymongo import MongoClient from pymongo.database import Database @@ -17,7 +17,13 @@ from functools import cached_property -MongoVersion: TypeAlias = Tuple[int, int] +try: + from typing import TypeAlias # pylint: disable=ungrouped-imports + + MongoVersion: TypeAlias = Tuple[int, int] +except ImportError: + TypeAlias = None + MongoVersion = Tuple[int, int] class MongoDBConnector: From 61897349b2f0b97454506b7c7587d84d576e2af1 Mon Sep 17 00:00:00 2001 From: Matt Menzenski Date: Wed, 18 Oct 2023 08:39:13 -0500 Subject: [PATCH 3/3] lock plugins --- plugins/loaders/target-jsonl--andyh1203.lock | 34 ++++++++++++++++++++ 1 file changed, 34 insertions(+) create mode 100644 plugins/loaders/target-jsonl--andyh1203.lock diff --git a/plugins/loaders/target-jsonl--andyh1203.lock b/plugins/loaders/target-jsonl--andyh1203.lock new file mode 100644 index 0000000..11fa0ba --- /dev/null +++ b/plugins/loaders/target-jsonl--andyh1203.lock @@ -0,0 +1,34 @@ +{ + "plugin_type": "loaders", + "name": "target-jsonl", + "namespace": "target_jsonl", + "variant": "andyh1203", + "label": "JSON Lines (JSONL)", + "docs": "https://hub.meltano.com/loaders/target-jsonl--andyh1203", + "repo": "https://github.com/andyh1203/target-jsonl", + "pip_url": "target-jsonl", + "description": "JSONL loader", + "logo_url": "https://hub.meltano.com/assets/logos/loaders/jsonl.png", + "settings": [ + { + "name": "destination_path", + "kind": "string", + "value": "output", + "label": "Destination Path", + "description": "Sets the destination path the JSONL files are written to, relative\nto the project root.\n\nThe directory needs to exist already, it will not be created\nautomatically.\n\nTo write JSONL files to the project root, set an empty string (`\"\"`).\n" + }, + { + "name": "do_timestamp_file", + "kind": "boolean", + "value": false, + "label": "Include Timestamp in File Names", + "description": "Specifies if the files should get timestamped.\n\nBy default, the resulting file will not have a timestamp in the file name (i.e. `exchange_rate.jsonl`).\n\nIf this option gets set to `true`, the resulting file will have a timestamp associated with it (i.e. `exchange_rate-{timestamp}.jsonl`).\n" + }, + { + "name": "custom_name", + "kind": "string", + "label": "Custom File Name Override", + "description": "Specifies a custom name for the filename, instead of the stream name.\n\nThe file name will be `{custom_name}-{timestamp}.jsonl`, if `do_timestamp_file` is `true`.\nOtherwise the file name will be `{custom_name}.jsonl`.\n\nIf custom name is not provided, the stream name will be used.\n" + } + ] +}