diff --git a/aws_lambda_powertools/utilities/idempotency/base.py b/aws_lambda_powertools/utilities/idempotency/base.py
index a8d509b86eb..771547fe33c 100644
--- a/aws_lambda_powertools/utilities/idempotency/base.py
+++ b/aws_lambda_powertools/utilities/idempotency/base.py
@@ -14,8 +14,10 @@
     IdempotencyValidationError,
 )
 from aws_lambda_powertools.utilities.idempotency.persistence.base import (
-    STATUS_CONSTANTS,
     BasePersistenceLayer,
+)
+from aws_lambda_powertools.utilities.idempotency.persistence.datarecord import (
+    STATUS_CONSTANTS,
     DataRecord,
 )
 from aws_lambda_powertools.utilities.idempotency.serialization.base import (
@@ -118,12 +120,17 @@ def _process_idempotency(self):
                 data=self.data,
                 remaining_time_in_millis=self._get_remaining_time_in_millis(),
             )
-        except IdempotencyKeyError:
+        except (IdempotencyKeyError, IdempotencyValidationError):
             raise
-        except IdempotencyItemAlreadyExistsError:
-            # Now we know the item already exists, we can retrieve it
-            record = self._get_idempotency_record()
-            if record is not None:
+        except IdempotencyItemAlreadyExistsError as exc:
+            # Attempt to retrieve the existing record, either from the exception ReturnValuesOnConditionCheckFailure
+            # or perform a GET operation if the information is not available.
+            # We give preference to ReturnValuesOnConditionCheckFailure because it is a faster and more cost-effective
+            # way of retrieving the existing record after a failed conditional write operation.
+            record = exc.old_data_record or self._get_idempotency_record()
+
+            # If a record is found, handle it for status
+            if record:
                 return self._handle_for_status(record)
         except Exception as exc:
             raise IdempotencyPersistenceLayerError(
diff --git a/aws_lambda_powertools/utilities/idempotency/exceptions.py b/aws_lambda_powertools/utilities/idempotency/exceptions.py
index e4c57a8f2b6..55ec662799e 100644
--- a/aws_lambda_powertools/utilities/idempotency/exceptions.py
+++ b/aws_lambda_powertools/utilities/idempotency/exceptions.py
@@ -5,6 +5,8 @@
 
 from typing import Optional, Union
 
+from aws_lambda_powertools.utilities.idempotency.persistence.datarecord import DataRecord
+
 
 class BaseError(Exception):
     """
@@ -30,6 +32,18 @@ class IdempotencyItemAlreadyExistsError(BaseError):
     Item attempting to be inserted into persistence store already exists and is not expired
     """
 
+    def __init__(self, *args: Optional[Union[str, Exception]], old_data_record: Optional[DataRecord] = None):
+        self.old_data_record = old_data_record
+        super().__init__(*args)
+
+    def __str__(self):
+        """
+        Return all arguments formatted or original message
+        """
+        old_data_record = f" from [{(str(self.old_data_record))}]" if self.old_data_record else ""
+        message = super().__str__()
+        return f"{message}{old_data_record}"
+
 
 class IdempotencyItemNotFoundError(BaseError):
     """
diff --git a/aws_lambda_powertools/utilities/idempotency/persistence/base.py b/aws_lambda_powertools/utilities/idempotency/persistence/base.py
index f3b12da0310..335c7ecc9fb 100644
--- a/aws_lambda_powertools/utilities/idempotency/persistence/base.py
+++ b/aws_lambda_powertools/utilities/idempotency/persistence/base.py
@@ -8,8 +8,7 @@
 import os
 import warnings
 from abc import ABC, abstractmethod
-from types import MappingProxyType
-from typing import Any, Dict, Optional
+from typing import Any, Dict, Optional, Union
 
 import jmespath
 
@@ -18,95 +17,18 @@
 from aws_lambda_powertools.shared.json_encoder import Encoder
 from aws_lambda_powertools.utilities.idempotency.config import IdempotencyConfig
 from aws_lambda_powertools.utilities.idempotency.exceptions import (
-    IdempotencyInvalidStatusError,
     IdempotencyItemAlreadyExistsError,
     IdempotencyKeyError,
     IdempotencyValidationError,
 )
+from aws_lambda_powertools.utilities.idempotency.persistence.datarecord import (
+    STATUS_CONSTANTS,
+    DataRecord,
+)
 from aws_lambda_powertools.utilities.jmespath_utils import PowertoolsFunctions
 
 logger = logging.getLogger(__name__)
 
-STATUS_CONSTANTS = MappingProxyType({"INPROGRESS": "INPROGRESS", "COMPLETED": "COMPLETED", "EXPIRED": "EXPIRED"})
-
-
-class DataRecord:
-    """
-    Data Class for idempotency records.
-    """
-
-    def __init__(
-        self,
-        idempotency_key: str,
-        status: str = "",
-        expiry_timestamp: Optional[int] = None,
-        in_progress_expiry_timestamp: Optional[int] = None,
-        response_data: str = "",
-        payload_hash: str = "",
-    ) -> None:
-        """
-
-        Parameters
-        ----------
-        idempotency_key: str
-            hashed representation of the idempotent data
-        status: str, optional
-            status of the idempotent record
-        expiry_timestamp: int, optional
-            time before the record should expire, in seconds
-        in_progress_expiry_timestamp: int, optional
-            time before the record should expire while in the INPROGRESS state, in seconds
-        payload_hash: str, optional
-            hashed representation of payload
-        response_data: str, optional
-            response data from previous executions using the record
-        """
-        self.idempotency_key = idempotency_key
-        self.payload_hash = payload_hash
-        self.expiry_timestamp = expiry_timestamp
-        self.in_progress_expiry_timestamp = in_progress_expiry_timestamp
-        self._status = status
-        self.response_data = response_data
-
-    @property
-    def is_expired(self) -> bool:
-        """
-        Check if data record is expired
-
-        Returns
-        -------
-        bool
-            Whether the record is currently expired or not
-        """
-        return bool(self.expiry_timestamp and int(datetime.datetime.now().timestamp()) > self.expiry_timestamp)
-
-    @property
-    def status(self) -> str:
-        """
-        Get status of data record
-
-        Returns
-        -------
-        str
-        """
-        if self.is_expired:
-            return STATUS_CONSTANTS["EXPIRED"]
-        elif self._status in STATUS_CONSTANTS.values():
-            return self._status
-        else:
-            raise IdempotencyInvalidStatusError(self._status)
-
-    def response_json_as_dict(self) -> Optional[dict]:
-        """
-        Get response data deserialized to python dict
-
-        Returns
-        -------
-        Optional[dict]
-            previous response data deserialized
-        """
-        return json.loads(self.response_data) if self.response_data else None
-
 
 class BasePersistenceLayer(ABC):
     """
@@ -238,16 +160,20 @@ def _generate_hash(self, data: Any) -> str:
         hashed_data = self.hash_function(json.dumps(data, cls=Encoder, sort_keys=True).encode())
         return hashed_data.hexdigest()
 
-    def _validate_payload(self, data: Dict[str, Any], data_record: DataRecord) -> None:
+    def _validate_payload(
+        self,
+        data_payload: Union[Dict[str, Any], DataRecord],
+        stored_data_record: DataRecord,
+    ) -> None:
         """
         Validate that the hashed payload matches data provided and stored data record
 
         Parameters
         ----------
-        data: Dict[str, Any]
+        data_payload: Union[Dict[str, Any], DataRecord]
             Payload
-        data_record: DataRecord
-            DataRecord instance
+        stored_data_record: DataRecord
+            DataRecord fetched from Dynamo or cache
 
         Raises
         ----------
@@ -256,8 +182,12 @@ def _validate_payload(self, data: Dict[str, Any], data_record: DataRecord) -> No
 
         """
         if self.payload_validation_enabled:
-            data_hash = self._get_hashed_payload(data=data)
-            if data_record.payload_hash != data_hash:
+            if isinstance(data_payload, DataRecord):
+                data_hash = data_payload.payload_hash
+            else:
+                data_hash = self._get_hashed_payload(data=data_payload)
+
+            if stored_data_record.payload_hash != data_hash:
                 raise IdempotencyValidationError("Payload does not match stored record for this event key")
 
     def _get_expiry_timestamp(self) -> int:
@@ -448,14 +378,14 @@ def get_record(self, data: Dict[str, Any]) -> Optional[DataRecord]:
         cached_record = self._retrieve_from_cache(idempotency_key=idempotency_key)
         if cached_record:
             logger.debug(f"Idempotency record found in cache with idempotency key: {idempotency_key}")
-            self._validate_payload(data=data, data_record=cached_record)
+            self._validate_payload(data_payload=data, stored_data_record=cached_record)
             return cached_record
 
         record = self._get_record(idempotency_key=idempotency_key)
 
         self._save_to_cache(data_record=record)
 
-        self._validate_payload(data=data, data_record=record)
+        self._validate_payload(data_payload=data, stored_data_record=record)
         return record
 
     @abstractmethod
diff --git a/aws_lambda_powertools/utilities/idempotency/persistence/datarecord.py b/aws_lambda_powertools/utilities/idempotency/persistence/datarecord.py
new file mode 100644
index 00000000000..607e238c3a0
--- /dev/null
+++ b/aws_lambda_powertools/utilities/idempotency/persistence/datarecord.py
@@ -0,0 +1,93 @@
+"""
+Data Class for idempotency records.
+"""
+
+import datetime
+import json
+import logging
+from types import MappingProxyType
+from typing import Optional
+
+logger = logging.getLogger(__name__)
+
+STATUS_CONSTANTS = MappingProxyType({"INPROGRESS": "INPROGRESS", "COMPLETED": "COMPLETED", "EXPIRED": "EXPIRED"})
+
+
+class DataRecord:
+    """
+    Data Class for idempotency records.
+    """
+
+    def __init__(
+        self,
+        idempotency_key: str,
+        status: str = "",
+        expiry_timestamp: Optional[int] = None,
+        in_progress_expiry_timestamp: Optional[int] = None,
+        response_data: str = "",
+        payload_hash: str = "",
+    ) -> None:
+        """
+
+        Parameters
+        ----------
+        idempotency_key: str
+            hashed representation of the idempotent data
+        status: str, optional
+            status of the idempotent record
+        expiry_timestamp: int, optional
+            time before the record should expire, in seconds
+        in_progress_expiry_timestamp: int, optional
+            time before the record should expire while in the INPROGRESS state, in seconds
+        payload_hash: str, optional
+            hashed representation of payload
+        response_data: str, optional
+            response data from previous executions using the record
+        """
+        self.idempotency_key = idempotency_key
+        self.payload_hash = payload_hash
+        self.expiry_timestamp = expiry_timestamp
+        self.in_progress_expiry_timestamp = in_progress_expiry_timestamp
+        self._status = status
+        self.response_data = response_data
+
+    @property
+    def is_expired(self) -> bool:
+        """
+        Check if data record is expired
+
+        Returns
+        -------
+        bool
+            Whether the record is currently expired or not
+        """
+        return bool(self.expiry_timestamp and int(datetime.datetime.now().timestamp()) > self.expiry_timestamp)
+
+    @property
+    def status(self) -> str:
+        """
+        Get status of data record
+
+        Returns
+        -------
+        str
+        """
+        if self.is_expired:
+            return STATUS_CONSTANTS["EXPIRED"]
+        if self._status in STATUS_CONSTANTS.values():
+            return self._status
+
+        from aws_lambda_powertools.utilities.idempotency.exceptions import IdempotencyInvalidStatusError
+
+        raise IdempotencyInvalidStatusError(self._status)
+
+    def response_json_as_dict(self) -> Optional[dict]:
+        """
+        Get response data deserialized to python dict
+
+        Returns
+        -------
+        Optional[dict]
+            previous response data deserialized
+        """
+        return json.loads(self.response_data) if self.response_data else None
diff --git a/aws_lambda_powertools/utilities/idempotency/persistence/dynamodb.py b/aws_lambda_powertools/utilities/idempotency/persistence/dynamodb.py
index 913e88524e2..6cb86121092 100644
--- a/aws_lambda_powertools/utilities/idempotency/persistence/dynamodb.py
+++ b/aws_lambda_powertools/utilities/idempotency/persistence/dynamodb.py
@@ -15,8 +15,9 @@
 from aws_lambda_powertools.utilities.idempotency.exceptions import (
     IdempotencyItemAlreadyExistsError,
     IdempotencyItemNotFoundError,
+    IdempotencyValidationError,
 )
-from aws_lambda_powertools.utilities.idempotency.persistence.base import (
+from aws_lambda_powertools.utilities.idempotency.persistence.datarecord import (
     STATUS_CONSTANTS,
     DataRecord,
 )
@@ -114,6 +115,14 @@ def __init__(
         self.data_attr = data_attr
         self.validation_key_attr = validation_key_attr
 
+        # Use DynamoDB's ReturnValuesOnConditionCheckFailure to optimize put and get operations and optimize costs.
+        # This feature is supported in boto3 versions 1.26.164 and later.
+        self.return_value_on_condition = (
+            {"ReturnValuesOnConditionCheckFailure": "ALL_OLD"}
+            if self.boto3_supports_condition_check_failure(boto3.__version__)
+            else {}
+        )
+
         self._deserializer = TypeDeserializer()
 
         super(DynamoDBPersistenceLayer, self).__init__()
@@ -221,6 +230,7 @@ def _put_record(self, data_record: DataRecord) -> None:
             condition_expression = (
                 f"{idempotency_key_not_exist} OR {idempotency_expiry_expired} OR ({inprogress_expiry_expired})"
             )
+
             self.client.put_item(
                 TableName=self.table_name,
                 Item=item,
@@ -236,16 +246,53 @@ def _put_record(self, data_record: DataRecord) -> None:
                     ":now_in_millis": {"N": str(int(now.timestamp() * 1000))},
                     ":inprogress": {"S": STATUS_CONSTANTS["INPROGRESS"]},
                 },
+                **self.return_value_on_condition,  # type: ignore
             )
         except ClientError as exc:
             error_code = exc.response.get("Error", {}).get("Code")
             if error_code == "ConditionalCheckFailedException":
+                old_data_record = self._item_to_data_record(exc.response["Item"]) if "Item" in exc.response else None
+                if old_data_record is not None:
+                    logger.debug(
+                        f"Failed to put record for already existing idempotency key: "
+                        f"{data_record.idempotency_key} with status: {old_data_record.status}, "
+                        f"expiry_timestamp: {old_data_record.expiry_timestamp}, "
+                        f"and in_progress_expiry_timestamp: {old_data_record.in_progress_expiry_timestamp}",
+                    )
+                    self._save_to_cache(data_record=old_data_record)
+
+                    try:
+                        self._validate_payload(data_payload=data_record, stored_data_record=old_data_record)
+                    except IdempotencyValidationError as idempotency_validation_error:
+                        raise idempotency_validation_error from exc
+
+                    raise IdempotencyItemAlreadyExistsError(old_data_record=old_data_record) from exc
+
                 logger.debug(
                     f"Failed to put record for already existing idempotency key: {data_record.idempotency_key}",
                 )
-                raise IdempotencyItemAlreadyExistsError from exc
-            else:
-                raise
+                raise IdempotencyItemAlreadyExistsError() from exc
+
+            raise
+
+    @staticmethod
+    def boto3_supports_condition_check_failure(boto3_version: str) -> bool:
+        """
+        Check if the installed boto3 version supports condition check failure.
+
+        Params
+        ------
+        boto3_version: str
+            The boto3 version
+
+        Returns
+        -------
+        bool
+            True if the boto3 version supports condition check failure, False otherwise.
+        """
+        # Only supported in boto3 1.26.164 and above
+        major, minor, *patch = map(int, boto3_version.split("."))
+        return (major, minor, *patch) >= (1, 26, 164)
 
     def _update_record(self, data_record: DataRecord):
         logger.debug(f"Updating record for idempotency key: {data_record.idempotency_key}")
diff --git a/docs/media/idempotency_first_execution.png b/docs/media/idempotency_first_execution.png
new file mode 100644
index 00000000000..23d185bbf6f
Binary files /dev/null and b/docs/media/idempotency_first_execution.png differ
diff --git a/docs/media/idempotency_second_execution.png b/docs/media/idempotency_second_execution.png
new file mode 100644
index 00000000000..32ecb13ef0f
Binary files /dev/null and b/docs/media/idempotency_second_execution.png differ
diff --git a/docs/utilities/idempotency.md b/docs/utilities/idempotency.md
index 3b7fe344b1c..bc355580935 100644
--- a/docs/utilities/idempotency.md
+++ b/docs/utilities/idempotency.md
@@ -102,12 +102,16 @@ If you're not [changing the default configuration for the DynamoDB persistence l
 
     Larger items cannot be written to DynamoDB and will cause exceptions. If your response exceeds 400kb, consider using Redis as your persistence layer.
 
+<!-- markdownlint-disable MD013 -->
 ???+ info "Info: DynamoDB"
-    Each function invocation will generally make 2 requests to DynamoDB. If the
-    result returned by your Lambda is less than 1kb, you can expect 2 WCUs per invocation. For retried invocations, you will
-    see 1WCU and 1RCU. Review the [DynamoDB pricing documentation](https://aws.amazon.com/dynamodb/pricing/){target="_blank"} to
-    estimate the cost.
 
+    During the first invocation with a payload, the Lambda function executes both a `PutItem` and an `UpdateItem` operations to store the data in DynamoDB. If the result returned by your Lambda is less than 1kb, you can expect 2 WCUs per Lambda invocation.
+
+    On subsequent invocations with the same payload, you can expect just 1 `PutItem` request to DynamoDB.
+
+    **Note:** While we try to minimize requests to DynamoDB to 1 per invocation, if your boto3 version is lower than `1.26.194`, you may experience 2 requests in every invocation. Ensure to check your boto3 version and review the [DynamoDB pricing documentation](https://aws.amazon.com/dynamodb/pricing/){target="_blank"} to estimate the cost.
+
+<!-- markdownlint-enable MD013 -->
 ### Idempotent decorator
 
 You can quickly start by initializing the `DynamoDBPersistenceLayer` class and using it with the `idempotent` decorator on your lambda handler.
@@ -936,6 +940,24 @@ The idempotency utility can be used with the `validator` decorator. Ensure that
 ???+ tip "Tip: JMESPath Powertools for AWS Lambda (Python) functions are also available"
     Built-in functions known in the validation utility like `powertools_json`, `powertools_base64`, `powertools_base64_gzip` are also available to use in this utility.
 
+### Tracer
+
+The idempotency utility can be used with the `tracer` decorator. Ensure that idempotency is the innermost decorator.
+
+#### First execution
+
+During the first execution with a payload, Lambda performs a `PutItem` followed by an `UpdateItem` operation to persist the record in DynamoDB.
+
+![Tracer showcase](../media/idempotency_first_execution.png)
+
+#### Subsequent executions
+
+On subsequent executions with the same payload, Lambda optimistically tries to save the record in DynamoDB. If the record already exists, DynamoDB returns the item.
+
+Explore how to handle conditional write errors in high-concurrency scenarios with DynamoDB in this [blog post](https://aws.amazon.com/pt/blogs/database/handle-conditional-write-errors-in-high-concurrency-scenarios-with-amazon-dynamodb/){target="_blank"}.
+
+![Tracer showcase](../media/idempotency_second_execution.png)
+
 ## Testing your code
 
 The idempotency utility provides several routes to test your code.
diff --git a/tests/functional/idempotency/conftest.py b/tests/functional/idempotency/conftest.py
index 9bacfb58779..f8d48cd7da2 100644
--- a/tests/functional/idempotency/conftest.py
+++ b/tests/functional/idempotency/conftest.py
@@ -131,6 +131,7 @@ def expected_params_put_item(hashed_idempotency_key):
             "attribute_not_exists(#id) OR #expiry < :now OR "
             "(#status = :inprogress AND attribute_exists(#in_progress_expiry) AND #in_progress_expiry < :now_in_millis)"
         ),
+        "ReturnValuesOnConditionCheckFailure": "ALL_OLD",
         "ExpressionAttributeNames": {
             "#id": "id",
             "#expiry": "expiration",
@@ -159,6 +160,7 @@ def expected_params_put_item_with_validation(hashed_idempotency_key, hashed_vali
             "attribute_not_exists(#id) OR #expiry < :now OR "
             "(#status = :inprogress AND attribute_exists(#in_progress_expiry) AND #in_progress_expiry < :now_in_millis)"
         ),
+        "ReturnValuesOnConditionCheckFailure": "ALL_OLD",
         "ExpressionAttributeNames": {
             "#id": "id",
             "#expiry": "expiration",
diff --git a/tests/functional/idempotency/test_idempotency.py b/tests/functional/idempotency/test_idempotency.py
index 24fcd76b4d5..905248011e6 100644
--- a/tests/functional/idempotency/test_idempotency.py
+++ b/tests/functional/idempotency/test_idempotency.py
@@ -87,14 +87,7 @@ def test_idempotent_lambda_already_completed(
             "status": {"S": "COMPLETED"},
         },
     }
-
-    expected_params = {
-        "TableName": TABLE_NAME,
-        "Key": {"id": {"S": hashed_idempotency_key}},
-        "ConsistentRead": True,
-    }
-    stubber.add_client_error("put_item", "ConditionalCheckFailedException")
-    stubber.add_response("get_item", ddb_response, expected_params)
+    stubber.add_client_error("put_item", "ConditionalCheckFailedException", modeled_fields=ddb_response)
     stubber.activate()
 
     @idempotent(config=idempotency_config, persistence_store=persistence_store)
@@ -124,11 +117,6 @@ def test_idempotent_lambda_in_progress(
 
     stubber = stub.Stubber(persistence_store.client)
 
-    expected_params = {
-        "TableName": TABLE_NAME,
-        "Key": {"id": {"S": hashed_idempotency_key}},
-        "ConsistentRead": True,
-    }
     ddb_response = {
         "Item": {
             "id": {"S": hashed_idempotency_key},
@@ -137,8 +125,7 @@ def test_idempotent_lambda_in_progress(
         },
     }
 
-    stubber.add_client_error("put_item", "ConditionalCheckFailedException")
-    stubber.add_response("get_item", ddb_response, expected_params)
+    stubber.add_client_error("put_item", "ConditionalCheckFailedException", modeled_fields=ddb_response)
     stubber.activate()
 
     @idempotent(config=idempotency_config, persistence_store=persistence_store)
@@ -176,11 +163,6 @@ def test_idempotent_lambda_in_progress_with_cache(
     retrieve_from_cache_spy = mocker.spy(persistence_store, "_retrieve_from_cache")
     stubber = stub.Stubber(persistence_store.client)
 
-    expected_params = {
-        "TableName": TABLE_NAME,
-        "Key": {"id": {"S": hashed_idempotency_key}},
-        "ConsistentRead": True,
-    }
     ddb_response = {
         "Item": {
             "id": {"S": hashed_idempotency_key},
@@ -189,14 +171,12 @@ def test_idempotent_lambda_in_progress_with_cache(
         },
     }
 
-    stubber.add_client_error("put_item", "ConditionalCheckFailedException")
-    stubber.add_response("get_item", ddb_response, expected_params)
+    stubber.add_client_error("put_item", "ConditionalCheckFailedException", modeled_fields=ddb_response)
 
-    stubber.add_client_error("put_item", "ConditionalCheckFailedException")
-    stubber.add_response("get_item", copy.deepcopy(ddb_response), copy.deepcopy(expected_params))
+    stubber.add_client_error("put_item", "ConditionalCheckFailedException", modeled_fields=copy.deepcopy(ddb_response))
+
+    stubber.add_client_error("put_item", "ConditionalCheckFailedException", modeled_fields=copy.deepcopy(ddb_response))
 
-    stubber.add_client_error("put_item", "ConditionalCheckFailedException")
-    stubber.add_response("get_item", copy.deepcopy(ddb_response), copy.deepcopy(expected_params))
     stubber.activate()
 
     @idempotent(config=idempotency_config, persistence_store=persistence_store)
@@ -212,7 +192,7 @@ def lambda_handler(event, context):
                 "body=a3edd699125517bb49d562501179ecbd"
             )
 
-    assert retrieve_from_cache_spy.call_count == 2 * loops
+    assert retrieve_from_cache_spy.call_count == loops
     retrieve_from_cache_spy.assert_called_with(idempotency_key=hashed_idempotency_key)
 
     save_to_cache_spy.assert_called()
@@ -411,7 +391,6 @@ def lambda_handler(event, context):
     "idempotency_config",
     [
         {"use_local_cache": False, "payload_validation_jmespath": "requestContext"},
-        {"use_local_cache": True, "payload_validation_jmespath": "requestContext"},
     ],
     indirect=True,
 )
@@ -439,11 +418,7 @@ def test_idempotent_lambda_already_completed_with_validation_bad_payload(
             "validation": {"S": hashed_validation_key},
         },
     }
-
-    expected_params = {"TableName": TABLE_NAME, "Key": {"id": {"S": hashed_idempotency_key}}, "ConsistentRead": True}
-
-    stubber.add_client_error("put_item", "ConditionalCheckFailedException")
-    stubber.add_response("get_item", ddb_response, expected_params)
+    stubber.add_client_error("put_item", "ConditionalCheckFailedException", modeled_fields=ddb_response)
     stubber.activate()
 
     @idempotent(config=idempotency_config, persistence_store=persistence_store)
@@ -466,6 +441,7 @@ def test_idempotent_lambda_expired_during_request(
     timestamp_expired,
     lambda_response,
     hashed_idempotency_key,
+    mocker,
     lambda_context,
 ):
     """
@@ -502,6 +478,9 @@ def test_idempotent_lambda_expired_during_request(
 
     stubber.activate()
 
+    submethod_mocked = mocker.patch.object(persistence_store, "boto3_supports_condition_check_failure")
+    submethod_mocked.return_value = False
+
     @idempotent(config=idempotency_config, persistence_store=persistence_store)
     def lambda_handler(event, context):
         return lambda_response
@@ -514,6 +493,55 @@ def lambda_handler(event, context):
     stubber.deactivate()
 
 
+@pytest.mark.parametrize(
+    "idempotency_config",
+    [{"use_local_cache": True, "payload_validation_jmespath": "requestContext"}],
+    indirect=True,
+)
+def test_idempotent_lambda_already_completed_with_validation_bad_payload_from_local_cache(
+    idempotency_config: IdempotencyConfig,
+    persistence_store: DynamoDBPersistenceLayer,
+    lambda_apigw_event,
+    timestamp_future,
+    lambda_response,
+    hashed_idempotency_key,
+    hashed_validation_key,
+    lambda_context,
+):
+    """
+    Test idempotent decorator where event with matching event key has already been successfully processed
+    Fetching record from local cache
+    """
+
+    stubber = stub.Stubber(persistence_store.client)
+    ddb_response = {
+        "Item": {
+            "id": {"S": hashed_idempotency_key},
+            "expiration": {"N": timestamp_future},
+            "data": {"S": '{"message": "test", "statusCode": 200}'},
+            "status": {"S": "COMPLETED"},
+            "validation": {"S": hashed_validation_key},
+        },
+    }
+
+    expected_params = {"TableName": TABLE_NAME, "Key": {"id": {"S": hashed_idempotency_key}}, "ConsistentRead": True}
+
+    stubber.add_client_error("put_item", "ConditionalCheckFailedException")
+    stubber.add_response("get_item", ddb_response, expected_params)
+    stubber.activate()
+
+    @idempotent(config=idempotency_config, persistence_store=persistence_store)
+    def lambda_handler(event, context):
+        return lambda_response
+
+    with pytest.raises(IdempotencyValidationError):
+        lambda_apigw_event["requestContext"]["accountId"] += "1"  # Alter the request payload
+        lambda_handler(lambda_apigw_event, lambda_context)
+
+    stubber.assert_no_pending_responses()
+    stubber.deactivate()
+
+
 @pytest.mark.parametrize("idempotency_config", [{"use_local_cache": False}, {"use_local_cache": True}], indirect=True)
 def test_idempotent_persistence_exception_deleting(
     idempotency_config: IdempotencyConfig,
@@ -675,13 +703,7 @@ def test_idempotent_lambda_with_validator_util(
         },
     }
 
-    expected_params = {
-        "TableName": TABLE_NAME,
-        "Key": {"id": {"S": hashed_idempotency_key_with_envelope}},
-        "ConsistentRead": True,
-    }
-    stubber.add_client_error("put_item", "ConditionalCheckFailedException")
-    stubber.add_response("get_item", ddb_response, expected_params)
+    stubber.add_client_error("put_item", "ConditionalCheckFailedException", modeled_fields=ddb_response)
     stubber.activate()
 
     @validator(envelope=envelopes.API_GATEWAY_HTTP)
@@ -1791,7 +1813,6 @@ def test_idempotent_lambda_compound_already_completed(
     """
 
     stubber = stub.Stubber(persistence_store_compound.client)
-    stubber.add_client_error("put_item", "ConditionalCheckFailedException")
     ddb_response = {
         "Item": {
             "id": {"S": "idempotency#"},
@@ -1801,13 +1822,7 @@ def test_idempotent_lambda_compound_already_completed(
             "status": {"S": "COMPLETED"},
         },
     }
-    expected_params = {
-        "TableName": TABLE_NAME,
-        "Key": {"id": {"S": "idempotency#"}, "sk": {"S": hashed_idempotency_key}},
-        "ConsistentRead": True,
-    }
-    stubber.add_response("get_item", ddb_response, expected_params)
-
+    stubber.add_client_error("put_item", "ConditionalCheckFailedException", modeled_fields=ddb_response)
     stubber.activate()
 
     @idempotent(config=idempotency_config, persistence_store=persistence_store_compound)
diff --git a/tests/functional/idempotency/utils.py b/tests/functional/idempotency/utils.py
index bf57259fe76..60f28b075b6 100644
--- a/tests/functional/idempotency/utils.py
+++ b/tests/functional/idempotency/utils.py
@@ -26,6 +26,7 @@ def build_idempotency_put_item_stub(
             "attribute_not_exists(#id) OR #expiry < :now OR "
             "(#status = :inprogress AND attribute_exists(#in_progress_expiry) AND #in_progress_expiry < :now_in_millis)"
         ),
+        "ReturnValuesOnConditionCheckFailure": "ALL_OLD",
         "ExpressionAttributeNames": {
             "#id": "id",
             "#expiry": "expiration",
diff --git a/tests/unit/idempotency/test_dynamodb_persistence.py b/tests/unit/idempotency/test_dynamodb_persistence.py
index 9455c41ad8d..b27ef00550c 100644
--- a/tests/unit/idempotency/test_dynamodb_persistence.py
+++ b/tests/unit/idempotency/test_dynamodb_persistence.py
@@ -19,3 +19,14 @@ class DummyClient:
     # THEN
     assert persistence_layer.table_name == table_name
     assert persistence_layer.client == fake_client
+
+
+def test_boto3_version_supports_condition_check_failure():
+    assert DynamoDBPersistenceLayer.boto3_supports_condition_check_failure("0.0.3") is False
+    assert DynamoDBPersistenceLayer.boto3_supports_condition_check_failure("1.25") is False
+    assert DynamoDBPersistenceLayer.boto3_supports_condition_check_failure("1.25") is False
+    assert DynamoDBPersistenceLayer.boto3_supports_condition_check_failure("1.26.163") is False
+    assert DynamoDBPersistenceLayer.boto3_supports_condition_check_failure("1.26.164") is True
+    assert DynamoDBPersistenceLayer.boto3_supports_condition_check_failure("1.26.165") is True
+    assert DynamoDBPersistenceLayer.boto3_supports_condition_check_failure("1.27.0") is True
+    assert DynamoDBPersistenceLayer.boto3_supports_condition_check_failure("2.0.0") is True