[ESM] Add configurable poll frequency and log shard info (#12415)

gregfurman · web-flow · commit e70207f96f73 · 2025-03-20T15:15:50.000+02:00
diff --git a/localstack-core/localstack/config.py b/localstack-core/localstack/config.py
@@ -1071,6 +1071,24 @@ def populate_edge_configuration(
 # DEV: sbx_user1051 (default when not provided) Alternative system user or empty string to skip dropping privileges.
 LAMBDA_INIT_USER = os.environ.get("LAMBDA_INIT_USER")
 
+# INTERNAL: 1 (default)
+# The duration (in seconds) to wait between each poll call to an event source.
+LAMBDA_EVENT_SOURCE_MAPPING_POLL_INTERVAL_SEC = float(
+    os.environ.get("LAMBDA_EVENT_SOURCE_MAPPING_POLL_INTERVAL_SEC") or 1
+)
+
+# INTERNAL: 60 (default)
+# Maximum duration (in seconds) to wait between retries when an event source poll fails.
+LAMBDA_EVENT_SOURCE_MAPPING_MAX_BACKOFF_ON_ERROR_SEC = float(
+    os.environ.get("LAMBDA_EVENT_SOURCE_MAPPING_MAX_BACKOFF_ON_ERROR_SEC") or 60
+)
+
+# INTERNAL: 10 (default)
+# Maximum duration (in seconds) to wait between polls when an event source returns empty results.
+LAMBDA_EVENT_SOURCE_MAPPING_MAX_BACKOFF_ON_EMPTY_POLL_SEC = float(
+    os.environ.get("LAMBDA_EVENT_SOURCE_MAPPING_MAX_BACKOFF_ON_EMPTY_POLL_SEC") or 10
+)
+
 # Adding Stepfunctions default port
 LOCAL_PORT_STEPFUNCTIONS = int(os.environ.get("LOCAL_PORT_STEPFUNCTIONS") or 8083)
 # Stepfunctions lambda endpoint override
diff --git a/localstack-core/localstack/services/lambda_/event_source_mapping/esm_worker.py b/localstack-core/localstack/services/lambda_/event_source_mapping/esm_worker.py
@@ -5,6 +5,11 @@
 from localstack.aws.api.lambda_ import (
     EventSourceMappingConfiguration,
 )
+from localstack.config import (
+    LAMBDA_EVENT_SOURCE_MAPPING_MAX_BACKOFF_ON_EMPTY_POLL_SEC,
+    LAMBDA_EVENT_SOURCE_MAPPING_MAX_BACKOFF_ON_ERROR_SEC,
+    LAMBDA_EVENT_SOURCE_MAPPING_POLL_INTERVAL_SEC,
+)
 from localstack.services.lambda_.event_source_mapping.pollers.poller import (
     EmptyPollResultsException,
     Poller,
@@ -14,11 +19,6 @@
 from localstack.utils.backoff import ExponentialBackoff
 from localstack.utils.threads import FuncThread
 
-POLL_INTERVAL_SEC: float = 1
-MAX_BACKOFF_POLL_EMPTY_SEC: float = 10
-MAX_BACKOFF_POLL_ERROR_SEC: float = 60
-
-
 LOG = logging.getLogger(__name__)
 
 
@@ -144,10 +144,15 @@ def poller_loop(self, *args, **kwargs):
             self.update_esm_state_in_store(EsmState.ENABLED)
             self.state_transition_reason = self.user_state_reason
 
-        error_boff = ExponentialBackoff(initial_interval=2, max_interval=MAX_BACKOFF_POLL_ERROR_SEC)
-        empty_boff = ExponentialBackoff(initial_interval=1, max_interval=MAX_BACKOFF_POLL_EMPTY_SEC)
+        error_boff = ExponentialBackoff(
+            initial_interval=2, max_interval=LAMBDA_EVENT_SOURCE_MAPPING_MAX_BACKOFF_ON_ERROR_SEC
+        )
+        empty_boff = ExponentialBackoff(
+            initial_interval=1,
+            max_interval=LAMBDA_EVENT_SOURCE_MAPPING_MAX_BACKOFF_ON_EMPTY_POLL_SEC,
+        )
 
-        poll_interval_duration = POLL_INTERVAL_SEC
+        poll_interval_duration = LAMBDA_EVENT_SOURCE_MAPPING_POLL_INTERVAL_SEC
 
         while not self._shutdown_event.is_set():
             try:
@@ -159,7 +164,7 @@ def poller_loop(self, *args, **kwargs):
                 empty_boff.reset()
 
                 # Set the poll frequency back to the default
-                poll_interval_duration = POLL_INTERVAL_SEC
+                poll_interval_duration = LAMBDA_EVENT_SOURCE_MAPPING_POLL_INTERVAL_SEC
             except EmptyPollResultsException as miss_ex:
                 # If the event source is empty, backoff
                 poll_interval_duration = empty_boff.next_backoff()
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
@@ -135,6 +135,12 @@ def poll_events(self):
         if not self.shards:
             self.shards = self.initialize_shards()
 
+        if not self.shards:
+            LOG.debug("No shards found for %s.", self.source_arn)
+            raise EmptyPollResultsException(service=self.event_source(), source_arn=self.source_arn)
+        else:
+            LOG.debug("Event source %s has %d shards.", self.source_arn, len(self.shards))
+
         # TODO: improve efficiency because this currently limits the throughput to at most batch size per poll interval
         # Handle shards round-robin. Re-initialize current shard iterator once all shards are handled.
         if self.iterator_over_shards is None:
@@ -163,8 +169,11 @@ def poll_events_from_shard(self, shard_id: str, shard_iterator: str):
         get_records_response = self.get_records(shard_iterator)
         records = get_records_response.get("Records", [])
         if not records:
+            # We cannot reliably back-off when no records found since an iterator
+            # may have to move multiple times until records are returned.
+            # See https://docs.aws.amazon.com/streams/latest/dev/troubleshooting-consumers.html#getrecords-returns-empty
             self.shards[shard_id] = get_records_response["NextShardIterator"]
-            raise EmptyPollResultsException(service=self.event_source(), source_arn=self.source_arn)
+            return
 
         polled_events = self.transform_into_events(records, shard_id)
 
