1616
1717import static software .amazon .lambda .powertools .common .internal .LambdaConstants .LAMBDA_FUNCTION_NAME_ENV ;
1818
19- import com .fasterxml .jackson .core .JsonProcessingException ;
20- import com .fasterxml .jackson .databind .JsonNode ;
21- import com .fasterxml .jackson .databind .ObjectWriter ;
22- import io .burt .jmespath .Expression ;
2319import java .math .BigInteger ;
2420import java .nio .charset .StandardCharsets ;
2521import java .security .MessageDigest ;
3329import java .util .Spliterators ;
3430import java .util .stream .Stream ;
3531import java .util .stream .StreamSupport ;
32+
3633import org .slf4j .Logger ;
3734import org .slf4j .LoggerFactory ;
35+
36+ import com .fasterxml .jackson .core .JsonProcessingException ;
37+ import com .fasterxml .jackson .databind .JsonNode ;
38+ import com .fasterxml .jackson .databind .ObjectWriter ;
39+
40+ import io .burt .jmespath .Expression ;
3841import software .amazon .lambda .powertools .idempotency .IdempotencyConfig ;
3942import software .amazon .lambda .powertools .idempotency .exceptions .IdempotencyItemAlreadyExistsException ;
4043import software .amazon .lambda .powertools .idempotency .exceptions .IdempotencyItemNotFoundException ;
@@ -125,8 +128,7 @@ public void saveSuccess(JsonNode data, Object result, Instant now) {
125128 DataRecord .Status .COMPLETED ,
126129 getExpiryEpochSecond (now ),
127130 responseJson ,
128- getHashedPayload (data )
129- );
131+ getHashedPayload (data ));
130132 LOG .debug ("Function successfully executed. Saving record to persistence store with idempotency key: {}" ,
131133 dataRecord .getIdempotencyKey ());
132134 updateRecord (dataRecord );
@@ -157,8 +159,8 @@ public void saveInProgress(JsonNode data, Instant now, OptionalInt remainingTime
157159
158160 OptionalLong inProgressExpirationMsTimestamp = OptionalLong .empty ();
159161 if (remainingTimeInMs .isPresent ()) {
160- inProgressExpirationMsTimestamp =
161- OptionalLong .of (now .plus (remainingTimeInMs .getAsInt (), ChronoUnit .MILLIS ).toEpochMilli ());
162+ inProgressExpirationMsTimestamp = OptionalLong
163+ .of (now .plus (remainingTimeInMs .getAsInt (), ChronoUnit .MILLIS ).toEpochMilli ());
162164 }
163165
164166 DataRecord dataRecord = new DataRecord (
@@ -167,10 +169,23 @@ public void saveInProgress(JsonNode data, Instant now, OptionalInt remainingTime
167169 getExpiryEpochSecond (now ),
168170 null ,
169171 getHashedPayload (data ),
170- inProgressExpirationMsTimestamp
171- );
172+ inProgressExpirationMsTimestamp );
172173 LOG .debug ("saving in progress record for idempotency key: {}" , dataRecord .getIdempotencyKey ());
173- putRecord (dataRecord , now );
174+
175+ try {
176+ putRecord (dataRecord , now );
177+ } catch (IdempotencyItemAlreadyExistsException iaee ) {
178+ // Similar to getRecord, we need to call validatePayload before returning a data record.
179+ // PR https://github.com/aws-powertools/powertools-lambda-java/pull/1821 introduced returning a data record
180+ // through IdempotencyItemAlreadyExistsException to save DynamoDB calls when using DDB as store.
181+ Optional <DataRecord > dr = iaee .getDataRecord ();
182+ if (dr .isPresent ()) {
183+ // throws IdempotencyValidationException if payload validation is enabled and failing
184+ validatePayload (data , dr .get ());
185+ }
186+
187+ throw iaee ;
188+ }
174189 }
175190
176191 /**
@@ -188,7 +203,7 @@ public void deleteRecord(JsonNode data, Throwable throwable) {
188203
189204 String idemPotencyKey = hashedIdempotencyKey .get ();
190205 LOG .debug ("Function raised an exception {}. " +
191- "Clearing in progress record in persistence store for idempotency key: {}" ,
206+ "Clearing in progress record in persistence store for idempotency key: {}" ,
192207 throwable .getClass (),
193208 idemPotencyKey );
194209
@@ -255,9 +270,9 @@ private Optional<String> getHashedIdempotencyKey(JsonNode data) {
255270
256271 private boolean isMissingIdemPotencyKey (JsonNode data ) {
257272 if (data .isContainerNode ()) {
258- Stream <JsonNode > stream =
259- StreamSupport . stream ( Spliterators .spliteratorUnknownSize (data .elements (), Spliterator .ORDERED ),
260- false );
273+ Stream <JsonNode > stream = StreamSupport . stream (
274+ Spliterators .spliteratorUnknownSize (data .elements (), Spliterator .ORDERED ),
275+ false );
261276 return stream .allMatch (JsonNode ::isNull );
262277 }
263278 return data .isNull ();
0 commit comments