diff --git a/opentelemetry-sdk/src/opentelemetry/sdk/metrics/_internal/exemplar/exemplar_reservoir.py b/opentelemetry-sdk/src/opentelemetry/sdk/metrics/_internal/exemplar/exemplar_reservoir.py
index afe2dcba38a..7d23152b702 100644
--- a/opentelemetry-sdk/src/opentelemetry/sdk/metrics/_internal/exemplar/exemplar_reservoir.py
+++ b/opentelemetry-sdk/src/opentelemetry/sdk/metrics/_internal/exemplar/exemplar_reservoir.py
@@ -5,6 +5,7 @@
 from collections import defaultdict
 from collections.abc import Callable, Mapping, Sequence
 from random import randrange
+from threading import Lock
 from typing import (
     Any,
 )
@@ -151,6 +152,14 @@ def __init__(self, size: int, **kwargs) -> None:
         self._reservoir_storage: Mapping[int, ExemplarBucket] = defaultdict(
             ExemplarBucket
         )
+        # Guards `_reservoir_storage` and the subclass-specific reservoir state
+        # (e.g. `SimpleFixedSizeExemplarReservoir._measurements_seen`) against the
+        # offer()↔collect() race documented in #5431. Without it, collect() can
+        # run _reset() and zero `_measurements_seen` between the increment and the
+        # `randrange(0, self._measurements_seen)` call in `_find_bucket_index`,
+        # producing `ValueError: empty range for randrange() (0, 0, 0)`, and the
+        # `defaultdict` can also be read-while-written during collect()'s iteration.
+        self._lock: Lock = Lock()
 
     def collect(self, point_attributes: Attributes) -> list[Exemplar]:
         """Returns accumulated Exemplars and also resets the reservoir for the next
@@ -164,16 +173,17 @@ def collect(self, point_attributes: Attributes) -> list[Exemplar]:
             exemplars contain the attributes that were filtered out by the aggregator,
             but recorded alongside the original measurement.
         """
-        exemplars = [
-            e
-            for e in (
-                bucket.collect(point_attributes)
-                for _, bucket in sorted(self._reservoir_storage.items())
-            )
-            if e is not None
-        ]
-        self._reset()
-        return exemplars
+        with self._lock:
+            exemplars = [
+                e
+                for e in (
+                    bucket.collect(point_attributes)
+                    for _, bucket in sorted(self._reservoir_storage.items())
+                )
+                if e is not None
+            ]
+            self._reset()
+            return exemplars
 
     def offer(
         self,
@@ -190,17 +200,18 @@ def offer(
             attributes: Measurement attributes
             context: Measurement context
         """
-        try:
-            index = self._find_bucket_index(
-                value, time_unix_nano, attributes, context
-            )
-
-            self._reservoir_storage[index].offer(
-                value, time_unix_nano, attributes, context
-            )
-        except BucketIndexError:
-            # Ignore invalid bucket index
-            pass
+        with self._lock:
+            try:
+                index = self._find_bucket_index(
+                    value, time_unix_nano, attributes, context
+                )
+
+                self._reservoir_storage[index].offer(
+                    value, time_unix_nano, attributes, context
+                )
+            except BucketIndexError:
+                # Ignore invalid bucket index
+                pass
 
     @abstractmethod
     def _find_bucket_index(
@@ -287,12 +298,13 @@ def offer(
         context: Context,
     ) -> None:
         """Offers a measurement to be sampled."""
-        index = self._find_bucket_index(
-            value, time_unix_nano, attributes, context
-        )
-        self._reservoir_storage[index].offer(
-            value, time_unix_nano, attributes, context
-        )
+        with self._lock:
+            index = self._find_bucket_index(
+                value, time_unix_nano, attributes, context
+            )
+            self._reservoir_storage[index].offer(
+                value, time_unix_nano, attributes, context
+            )
 
     def _find_bucket_index(
         self,
diff --git a/opentelemetry-sdk/tests/metrics/test_exemplarreservoir.py b/opentelemetry-sdk/tests/metrics/test_exemplarreservoir.py
index 9c5bf69bdb3..bc845061680 100644
--- a/opentelemetry-sdk/tests/metrics/test_exemplarreservoir.py
+++ b/opentelemetry-sdk/tests/metrics/test_exemplarreservoir.py
@@ -1,7 +1,8 @@
 # Copyright The OpenTelemetry Authors
 # SPDX-License-Identifier: Apache-2.0
 
-from time import time_ns
+import threading
+from time import sleep, time_ns
 from unittest import TestCase
 
 from opentelemetry import trace
@@ -83,6 +84,66 @@ def test_reset_after_collection(self):
         self.assertEqual(new_exemplars[0].value, 4.0)
         self.assertEqual(new_exemplars[1].value, 5.0)
 
+    def test_concurrent_offer_and_collect_does_not_raise(self):
+        """Regression test for #5431.
+
+        ``offer()`` and ``collect()`` must be synchronized so that
+        ``collect()``→``_reset()`` cannot zero ``_measurements_seen`` between
+        the ``_measurements_seen += 1`` increment and the
+        ``randrange(0, self._measurements_seen)`` call in
+        ``_find_bucket_index``. Without the lock, that race produces
+        ``ValueError: empty range for randrange() (0, 0, 0)`` under concurrent
+        offer/collect.
+
+        size=1 maximizes the probability of hitting the randrange branch on
+        every offer after the first one, so the race shows up quickly.
+        """
+        reservoir = SimpleFixedSizeExemplarReservoir(size=1)
+        stop = threading.Event()
+        errors: list[BaseException] = []
+
+        def offer_loop() -> None:
+            i = 0
+            while not stop.is_set():
+                try:
+                    reservoir.offer(
+                        float(i), time_ns(), {"k": str(i)}, Context()
+                    )
+                    i += 1
+                except BaseException as e:  # noqa: BLE001 - surface any error
+                    errors.append(e)
+                    return
+
+        def collect_loop() -> None:
+            while not stop.is_set():
+                try:
+                    reservoir.collect({})
+                except BaseException as e:  # noqa: BLE001 - surface any error
+                    errors.append(e)
+                    return
+
+        threads = [
+            threading.Thread(target=offer_loop),
+            threading.Thread(target=offer_loop),
+            threading.Thread(target=collect_loop),
+        ]
+        for t in threads:
+            t.start()
+
+        # Bounded run. The race is reproducible in well under a second on
+        # Python 3.10+; 1.0s gives headroom on slow CI runners without
+        # noticeably slowing down the test suite.
+        sleep(1.0)
+        stop.set()
+        for t in threads:
+            t.join(timeout=5.0)
+
+        self.assertEqual(
+            errors,
+            [],
+            f"expected no errors under concurrent offer/collect, got: {errors}",
+        )
+
 
 class TestAlignedHistogramBucketExemplarReservoir(TestCase):
     TRACE_ID = int("d4cda95b652f4a1592b449d5929fda1b", 16)