fix: address Copilot review comments

- Defensive DD_DOGSTATSD_PORT parsing (catch ValueError, fallback to 8125)
- Wrap read loop in try/finally for final metrics snapshot on error
- Update emit_memory_metrics() to set _last_emission_time, avoiding
  duplicate initial snapshots
- Fix get_metrics_client() docstring to clarify caller must call initialize()

Co-Authored-By: bot_apk <apk@cognition.ai>

Author: devin-ai-integration[bot]

airbyte_cdk/entrypoint.py

@@ -283,12 +283,14 @@ def read(
 
         # The Airbyte protocol dictates that counts be expressed as float/double to better protect against integer overflows
         stream_message_counter: DefaultDict[HashableStreamDescriptor, float] = defaultdict(float)
-        for message in self.source.read(self.logger, config, catalog, state):
-            yield self.handle_record_counts(message, stream_message_counter)
-            # Periodically emit memory metrics (every 30s by default)
-            metrics_client.maybe_emit_memory_metrics()
-        # Emit final memory metrics snapshot
-        metrics_client.emit_memory_metrics()
+        try:
+            for message in self.source.read(self.logger, config, catalog, state):
+                yield self.handle_record_counts(message, stream_message_counter)
+                # Periodically emit memory metrics (every 30s by default)
+                metrics_client.maybe_emit_memory_metrics()
+        finally:
+            # Emit final memory metrics snapshot (runs even if the read loop raises)
+            metrics_client.emit_memory_metrics()
         for message in self._emit_queued_messages(self.source):
             yield self.handle_record_counts(message, stream_message_counter)
 

airbyte_cdk/metrics/__init__.py

@@ -56,7 +56,18 @@ def initialize(self) -> None:
             logger.debug("DD_AGENT_HOST not set; metrics emission disabled")
             return
 
-        dd_dogstatsd_port = int(os.environ.get("DD_DOGSTATSD_PORT", "8125"))
+        port_str = os.environ.get("DD_DOGSTATSD_PORT")
+        if not port_str:
+            dd_dogstatsd_port = 8125
+        else:
+            try:
+                dd_dogstatsd_port = int(port_str)
+            except ValueError:
+                logger.warning(
+                    "Invalid DD_DOGSTATSD_PORT value %r; falling back to default port 8125",
+                    port_str,
+                )
+                dd_dogstatsd_port = 8125
 
         try:
             from datadog.dogstatsd import DogStatsd
@@ -139,10 +150,16 @@ def emit_memory_metrics(self) -> None:
         - cdk.memory.usage_bytes: Current container memory usage
         - cdk.memory.limit_bytes: Container memory limit (if known)
         - cdk.memory.usage_percent: Usage/limit ratio (if limit is known)
+
+        Also updates the last-emission timestamp so that subsequent calls to
+        ``should_emit`` / ``maybe_emit_memory_metrics`` respect the interval.
         """
         if not self.enabled:
             return
 
+        # Update the last-emission timestamp to avoid duplicate snapshots
+        self._last_emission_time = time.monotonic()
+
         try:
             info: MemoryInfo = get_memory_info()
 
@@ -191,7 +208,10 @@ def get_metrics_client() -> MetricsClient:
     """
     Get or create the module-level MetricsClient singleton.
 
-    The client is initialized lazily on first access.
+    Note: The caller is responsible for calling ``initialize()`` on the
+    returned client before emitting metrics.  Construction and initialization
+    are separate so that the caller controls *when* environment variables
+    are read and the DogStatsD connection is established.
     """
     global _metrics_client
     if _metrics_client is None: