[ESM] Add backoff between stream poller retries

gregfurman · gregfurman · commit 05becffdcc28 · 2025-02-13T17:47:35.000+02:00
diff --git a/localstack-core/localstack/services/lambda_/event_source_mapping/pollers/stream_poller.py b/localstack-core/localstack/services/lambda_/event_source_mapping/pollers/stream_poller.py
@@ -1,5 +1,6 @@
 import json
 import logging
+import threading
 from abc import abstractmethod
 from datetime import datetime
 from typing import Iterator
@@ -28,6 +29,7 @@
 )
 from localstack.services.lambda_.event_source_mapping.pollers.sqs_poller import get_queue_url
 from localstack.utils.aws.arns import parse_arn, s3_bucket_name
+from localstack.utils.backoff import ExponentialBackoff
 from localstack.utils.strings import long_uid
 
 LOG = logging.getLogger(__name__)
@@ -47,6 +49,9 @@ class StreamPoller(Poller):
     # The ARN of the processor (e.g., Pipe ARN)
     partner_resource_arn: str | None
 
+    # Used for backing-off between retries and breaking the retry loop
+    _is_shutdown: threading.Event
+
     def __init__(
         self,
         source_arn: str,
@@ -62,6 +67,8 @@ def __init__(
         self.shards = {}
         self.iterator_over_shards = None
 
+        self._is_shutdown = threading.Event()
+
     @abstractmethod
     def transform_into_events(self, records: list[dict], shard_id) -> list[dict]:
         pass
@@ -104,12 +111,29 @@ def format_datetime(self, time: datetime) -> str:
     def get_sequence_number(self, record: dict) -> str:
         pass
 
+    def close(self):
+        self._is_shutdown.set()
+
     def pre_filter(self, events: list[dict]) -> list[dict]:
         return events
 
     def post_filter(self, events: list[dict]) -> list[dict]:
         return events
 
+    def has_record_expired(self, event: dict):
+        # Check MaximumRecordAgeInSeconds
+        if maximum_record_age_in_seconds := self.stream_parameters.get("MaximumRecordAgeInSeconds"):
+            arrival_timestamp_of_last_event = event.get("approximateArrivalTimestamp")
+            if not arrival_timestamp_of_last_event:
+                return False
+
+            now = get_current_time().timestamp()
+            record_age_in_seconds = now - arrival_timestamp_of_last_event
+            if record_age_in_seconds > maximum_record_age_in_seconds:
+                return True
+
+        return False
+
     def poll_events(self):
         """Generalized poller for streams such as Kinesis or DynamoDB
         Examples of Kinesis consumers:
@@ -146,14 +170,13 @@ def poll_events_from_shard(self, shard_id: str, shard_iterator: str):
         abort_condition = None
         get_records_response = self.get_records(shard_iterator)
         records = get_records_response["Records"]
+        if not records:
+            return
+
         polled_events = self.transform_into_events(records, shard_id)
         # Check MaximumRecordAgeInSeconds
-        if maximum_record_age_in_seconds := self.stream_parameters.get("MaximumRecordAgeInSeconds"):
-            arrival_timestamp_of_last_event = polled_events[-1]["approximateArrivalTimestamp"]
-            now = get_current_time().timestamp()
-            record_age_in_seconds = now - arrival_timestamp_of_last_event
-            if record_age_in_seconds > maximum_record_age_in_seconds:
-                abort_condition = "RecordAgeExpired"
+        if self.has_record_expired(polled_events[-1]):
+            abort_condition = "RecordAgeExpired"
 
         # TODO: implement format detection behavior (e.g., for JSON body):
         #  https://docs.aws.amazon.com/eventbridge/latest/userguide/eb-pipes-event-filtering.html
@@ -187,9 +210,26 @@ def poll_events_from_shard(self, shard_id: str, shard_iterator: str):
         # TODO: think about how to avoid starvation of other shards if one shard runs into infinite retries
         attempts = 0
         error_payload = {}
-        while not abort_condition and not self.max_retries_exceeded(attempts):
+
+        boff = ExponentialBackoff(max_retries=attempts)
+        while (
+            not abort_condition
+            and not self.max_retries_exceeded(attempts)
+            and not self._is_shutdown.is_set()
+        ):
             try:
+                if self.has_record_expired(polled_events[-1]):
+                    abort_condition = "RecordAgeExpired"
+
+                if attempts > 0:
+                    # TODO: Should we always backoff (with jitter) before processing since we may not want multiple pollers
+                    # all starting up and polling simultaneously
+                    # For example: 500 persisted ESMs starting up and requesting concurrently could flood gateway
+                    self._is_shutdown.wait(boff.next_backoff())
+
                 self.processor.process_events_batch(events)
+                boff.reset()
+
                 # Update shard iterator if execution is successful
                 self.shards[shard_id] = get_records_response["NextShardIterator"]
                 return
