Merge pull request #23 from menzenski/change-stream-start-after

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"
+    }
+  ]
+}

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"]

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
 
 from pymongo import MongoClient
 from pymongo.database import Database
@@ -17,6 +17,15 @@
     from functools import cached_property
 
 
+try:
+    from typing import TypeAlias  # pylint: disable=ungrouped-imports
+
+    MongoVersion: TypeAlias = Tuple[int, int]
+except ImportError:
+    TypeAlias = None
+    MongoVersion = Tuple[int, int]
+
+
 class MongoDBConnector:
     """MongoDB/DocumentDB connector class"""
 
@@ -34,6 +43,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 +52,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 +65,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,

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"]
                         ):