From 268db1d8ebf734f4f6cfcdd4df192b4d4824608e Mon Sep 17 00:00:00 2001 From: Arun Gopalpuri Date: Mon, 16 Feb 2026 13:07:32 -0800 Subject: [PATCH 1/6] idempotent rds module added --- README.md | 8 +- idempotent-core/pom.xml | 2 +- .../core/persistence/IdempotentStore.java | 6 +- .../persistence/InMemoryIdempotentStore.java | 20 +- idempotent-dynamo/pom.xml | 2 +- .../dynamo/DynamoIdempotentStore.java | 81 +++++-- idempotent-nats/pom.xml | 2 +- .../idempotent/nats/NatsIdempotentStore.java | 49 +++-- .../nats/NoopNatsAutoConfiguration.java | 8 +- idempotent-rds/README.md | 83 ++++++++ idempotent-rds/pom.xml | 102 +++++++++ .../idempotent/rds/RdsCleanupTask.java | 59 ++++++ .../arun0009/idempotent/rds/RdsConfig.java | 41 ++++ .../arun0009/idempotent/rds/RdsDialect.java | 32 +++ .../idempotent/rds/RdsIdempotentStore.java | 198 ++++++++++++++++++ ...ot.autoconfigure.AutoConfiguration.imports | 1 + .../rds/RdsIdempotentStoreH2Test.java | 126 +++++++++++ .../rds/RdsIdempotentStoreMySQLTest.java | 166 +++++++++++++++ .../rds/RdsIdempotentStorePostgreSQLTest.java | 158 ++++++++++++++ .../idempotent/rds/RdsMySQLTestConfig.java | 66 ++++++ .../rds/RdsPostgreSQLTestConfig.java | 66 ++++++ idempotent-redis/pom.xml | 2 +- .../redis/RedisIdempotentStore.java | 81 ++++++- pom.xml | 27 ++- 24 files changed, 1322 insertions(+), 64 deletions(-) create mode 100644 idempotent-rds/README.md create mode 100644 idempotent-rds/pom.xml create mode 100644 idempotent-rds/src/main/java/io/github/arun0009/idempotent/rds/RdsCleanupTask.java create mode 100644 idempotent-rds/src/main/java/io/github/arun0009/idempotent/rds/RdsConfig.java create mode 100644 idempotent-rds/src/main/java/io/github/arun0009/idempotent/rds/RdsDialect.java create mode 100644 idempotent-rds/src/main/java/io/github/arun0009/idempotent/rds/RdsIdempotentStore.java create mode 100644 idempotent-rds/src/main/resources/META-INF/spring/org.springframework.boot.autoconfigure.AutoConfiguration.imports create mode 100644 idempotent-rds/src/test/java/io/github/arun0009/idempotent/rds/RdsIdempotentStoreH2Test.java create mode 100644 idempotent-rds/src/test/java/io/github/arun0009/idempotent/rds/RdsIdempotentStoreMySQLTest.java create mode 100644 idempotent-rds/src/test/java/io/github/arun0009/idempotent/rds/RdsIdempotentStorePostgreSQLTest.java create mode 100644 idempotent-rds/src/test/java/io/github/arun0009/idempotent/rds/RdsMySQLTestConfig.java create mode 100644 idempotent-rds/src/test/java/io/github/arun0009/idempotent/rds/RdsPostgreSQLTestConfig.java diff --git a/README.md b/README.md index b9e2890..3e6d95b 100644 --- a/README.md +++ b/README.md @@ -2,7 +2,7 @@ Idempotent is a lightweight Java library that provides support for idempotency in APIs, making it easier to handle duplicate requests and ensuring reliable operation in distributed systems. This library integrates seamlessly with Spring applications -and offers idempotency support using Redis, DynamoDB, and NATS stores. +and offers idempotency support using Redis, DynamoDB, NATS, and RDS stores. Idempotent @@ -29,7 +29,7 @@ In API development, idempotency helps in the following ways: ## Features * **Two API Approaches**: Choose between annotation-based (AOP) or programmatic service-based idempotency depending on your needs. * **Integration with Spring**: Seamlessly integrates with Spring applications, providing annotations and utilities to easily add idempotency support to APIs. -* **Support for [Redis](idempotent-redis/README.md), [DynamoDB](idempotent-dynamo/README.md) or [NATS](idempotent-nats/README.md)**: Storage adapters for Redis, DynamoDB and NATS, allowing developers to choose the backend that best suits their requirements. +* **Support for [Redis](idempotent-redis/README.md), [DynamoDB](idempotent-dynamo/README.md), [NATS](idempotent-nats/README.md) or [RDS](idempotent-rds/README.md)**: Storage adapters for Redis, DynamoDB, NATS, and RDS, allowing developers to choose the backend that best suits their requirements. * **Simple Configuration**: Adding idempotency is as simple as annotating methods with [@Idempotent](idempotent-core/src/main/java/io/github/arun0009/idempotent/core/annotation/Idempotent.java) or using the `IdempotentService` programmatically. * **Client-Specified or Server-Specified Idempotent Keys**: Clients can dictate what the idempotent key should be via a configurable HTTP header, or the server can specify the idempotency key specified in the @Idempotent annotation configuration. * **Handling In-Progress Concurrent/Duplicate Requests**: Concurrent or duplicate requests will wait for the first request to complete (within a given configurable time frame and retries) and return the same response as the first request. @@ -38,8 +38,8 @@ In API development, idempotency helps in the following ways: ### Annotation-Based Approach (AOP) -1. Add the Idempotent maven dependency([redis](https://central.sonatype.com/artifact/io.github.arun0009/idempotent-redis), [dynamo](https://central.sonatype.com/artifact/io.github.arun0009/idempotent-dynamo) or [nats](https://central.sonatype.com/artifact/io.github.arun0009/idempotent-nats)) to your project. -2. Configure the storage backend ([Redis](idempotent-redis/README.md), [DynamoDB](idempotent-dynamo/README.md) or [NATS](idempotent-nats/README.md)) in your Spring application context. +1. Add the Idempotent maven dependency([redis](https://central.sonatype.com/artifact/io.github.arun0009/idempotent-redis), [dynamo](https://central.sonatype.com/artifact/io.github.arun0009/idempotent-dynamo), [nats](https://central.sonatype.com/artifact/io.github.arun0009/idempotent-nats) or [rds](https://central.sonatype.com/artifact/io.github.arun0009/idempotent-rds)) to your project. +2. Configure the storage backend ([Redis](idempotent-redis/README.md), [DynamoDB](idempotent-dynamo/README.md), [NATS](idempotent-nats/README.md) or [RDS](idempotent-rds/README.md)) in your Spring application context. 3. Annotate the desired API methods with [@Idempotent](idempotent-core/src/main/java/io/github/arun0009/idempotent/core/annotation/Idempotent.java): 4. Specify the key, time-to-live (TTL) as a Duration string, and/or if you want to hash the key for idempotent requests. diff --git a/idempotent-core/pom.xml b/idempotent-core/pom.xml index b78cd8d..b0e2753 100644 --- a/idempotent-core/pom.xml +++ b/idempotent-core/pom.xml @@ -5,7 +5,7 @@ io.github.arun0009 idempotent - 2.0.1 + 2.1.0 idempotent-core diff --git a/idempotent-core/src/main/java/io/github/arun0009/idempotent/core/persistence/IdempotentStore.java b/idempotent-core/src/main/java/io/github/arun0009/idempotent/core/persistence/IdempotentStore.java index 565eb6f..0cc8750 100644 --- a/idempotent-core/src/main/java/io/github/arun0009/idempotent/core/persistence/IdempotentStore.java +++ b/idempotent-core/src/main/java/io/github/arun0009/idempotent/core/persistence/IdempotentStore.java @@ -54,7 +54,11 @@ record IdempotentKey(String key, String processName) implements Serializable {} * @param expirationTimeInMilliSeconds expiry time of idempotent entry from store * @param response this is response received from downstream apis. */ - record Value(String status, Long expirationTimeInMilliSeconds, Object response) implements Serializable {} + record Value(String status, Long expirationTimeInMilliSeconds, Object response) implements Serializable { + public boolean isExpired() { + return expirationTimeInMilliSeconds != null && expirationTimeInMilliSeconds < System.currentTimeMillis(); + } + } /** * The enum Status. diff --git a/idempotent-core/src/main/java/io/github/arun0009/idempotent/core/persistence/InMemoryIdempotentStore.java b/idempotent-core/src/main/java/io/github/arun0009/idempotent/core/persistence/InMemoryIdempotentStore.java index dc10d82..48950da 100644 --- a/idempotent-core/src/main/java/io/github/arun0009/idempotent/core/persistence/InMemoryIdempotentStore.java +++ b/idempotent-core/src/main/java/io/github/arun0009/idempotent/core/persistence/InMemoryIdempotentStore.java @@ -23,7 +23,14 @@ public Value getValue(IdempotentKey idempotentKey, Class returnType) { @Override public void store(IdempotentKey idempotentKey, Value value) { - map.put(idempotentKey, value); + map.compute(idempotentKey, (k, v) -> { + if (v == null || v.isExpired()) { + return value; + } + // Race condition - key already exists and not expired + // Silent fail as per void interface contract + return v; + }); } @Override @@ -33,7 +40,16 @@ public void remove(IdempotentKey idempotentKey) { @Override public void update(IdempotentKey idempotentKey, Value value) { - map.replace(idempotentKey, value); + map.compute(idempotentKey, (k, v) -> { + if (v == null + || v.isExpired() + || IdempotentStore.Status.INPROGRESS.name().equals(v.status())) { + return value; + } + // Race condition - key not in correct state for update + // Silent fail as per void interface contract + return v; + }); } /** diff --git a/idempotent-dynamo/pom.xml b/idempotent-dynamo/pom.xml index 5f8336a..267c974 100644 --- a/idempotent-dynamo/pom.xml +++ b/idempotent-dynamo/pom.xml @@ -5,7 +5,7 @@ io.github.arun0009 idempotent - 2.0.1 + 2.1.0 idempotent-dynamo jar diff --git a/idempotent-dynamo/src/main/java/io/github/arun0009/idempotent/dynamo/DynamoIdempotentStore.java b/idempotent-dynamo/src/main/java/io/github/arun0009/idempotent/dynamo/DynamoIdempotentStore.java index c7fb2e9..bb2bf62 100644 --- a/idempotent-dynamo/src/main/java/io/github/arun0009/idempotent/dynamo/DynamoIdempotentStore.java +++ b/idempotent-dynamo/src/main/java/io/github/arun0009/idempotent/dynamo/DynamoIdempotentStore.java @@ -4,14 +4,20 @@ import io.github.arun0009.idempotent.core.persistence.IdempotentStore; import software.amazon.awssdk.enhanced.dynamodb.DynamoDbEnhancedClient; import software.amazon.awssdk.enhanced.dynamodb.DynamoDbTable; +import software.amazon.awssdk.enhanced.dynamodb.Expression; import software.amazon.awssdk.enhanced.dynamodb.Key; import software.amazon.awssdk.enhanced.dynamodb.TableSchema; import software.amazon.awssdk.enhanced.dynamodb.model.DeleteItemEnhancedRequest; +import software.amazon.awssdk.enhanced.dynamodb.model.PutItemEnhancedRequest; +import software.amazon.awssdk.services.dynamodb.model.AttributeValue; +import software.amazon.awssdk.services.dynamodb.model.ConditionalCheckFailedException; import tools.jackson.core.JacksonException; import tools.jackson.databind.json.JsonMapper; +import java.util.Map; + /** - * Dynamo idempotent store. + * Dynamo idempotent store with conditional expressions for race condition handling */ public class DynamoIdempotentStore implements IdempotentStore { @@ -48,10 +54,14 @@ public Value getValue(IdempotentKey idempotentKey, Class returnType) { if (idempotentItem == null) { return null; } - return new Value( + Value value = new Value( idempotentItem.getStatus(), idempotentItem.getExpirationTimeInMilliSeconds(), jsonMapper.readValue(idempotentItem.getResponse(), returnType)); + if (value.isExpired()) { + return null; + } + return value; } catch (JacksonException e) { throw new IdempotentException( String.format("Error getting response from dynamo for key: %s ", idempotentKey.key()), e); @@ -61,14 +71,24 @@ public Value getValue(IdempotentKey idempotentKey, Class returnType) { @Override public void store(IdempotentKey idempotentKey, Value value) { try { - DynamoDbTable idempotentTable = getTable(); - IdempotentItem idempotentItem = new IdempotentItem(); - idempotentItem.setKey(idempotentKey.key()); - idempotentItem.setProcessName(idempotentKey.processName()); - idempotentItem.setStatus(value.status()); - idempotentItem.setExpirationTimeInMilliSeconds(value.expirationTimeInMilliSeconds()); - idempotentItem.setResponse(jsonMapper.writeValueAsString(value.response())); - idempotentTable.putItem(idempotentItem); + IdempotentItem item = createItem(idempotentKey, value); + // Atomic condition: only insert if key doesn't exist or expired + Expression condition = Expression.builder() + .expression("attribute_not_exists(#key) OR expirationTimeInMilliSeconds < :now") + .expressionNames(Map.of("#key", "key")) + .expressionValues(Map.of( + ":now", + AttributeValue.builder() + .n(String.valueOf(System.currentTimeMillis())) + .build())) + .build(); + getTable() + .putItem(PutItemEnhancedRequest.builder(IdempotentItem.class) + .item(item) + .conditionExpression(condition) + .build()); + } catch (ConditionalCheckFailedException e) { + // Race condition - key exists and not expired, silent fail } catch (JacksonException e) { throw new IdempotentException("error storing idempotent item", e); } @@ -89,16 +109,41 @@ public void remove(IdempotentKey idempotentKey) { @Override public void update(IdempotentKey idempotentKey, Value value) { try { - DynamoDbTable idempotentTable = getTable(); - IdempotentItem idempotentItem = new IdempotentItem(); - idempotentItem.setKey(idempotentKey.key()); - idempotentItem.setProcessName(idempotentKey.processName()); - idempotentItem.setStatus(value.status()); - idempotentItem.setExpirationTimeInMilliSeconds(value.expirationTimeInMilliSeconds()); - idempotentItem.setResponse(jsonMapper.writeValueAsString(value.response())); - idempotentTable.putItem(idempotentItem); + IdempotentItem item = createItem(idempotentKey, value); + Expression condition = Expression.builder() + .expression( + "attribute_exists(#key) AND (#status = :inprogress OR expirationTimeInMilliSeconds < :now)") + .expressionNames(Map.of("#key", "key", "#status", "status")) + .expressionValues(Map.of( + ":now", + AttributeValue.builder() + .n(String.valueOf(System.currentTimeMillis())) + .build(), + ":inprogress", + AttributeValue.builder() + .s(IdempotentStore.Status.INPROGRESS.name()) + .build())) + .build(); + + getTable() + .putItem(PutItemEnhancedRequest.builder(IdempotentItem.class) + .item(item) + .conditionExpression(condition) + .build()); + } catch (ConditionalCheckFailedException e) { + // Race condition - key doesn't exist or not in correct state, silent fail } catch (JacksonException e) { throw new IdempotentException("error updating idempotent item", e); } } + + private IdempotentItem createItem(IdempotentKey idempotentKey, Value value) throws JacksonException { + IdempotentItem idempotentItem = new IdempotentItem(); + idempotentItem.setKey(idempotentKey.key()); + idempotentItem.setProcessName(idempotentKey.processName()); + idempotentItem.setStatus(value.status()); + idempotentItem.setExpirationTimeInMilliSeconds(value.expirationTimeInMilliSeconds()); + idempotentItem.setResponse(jsonMapper.writeValueAsString(value.response())); + return idempotentItem; + } } diff --git a/idempotent-nats/pom.xml b/idempotent-nats/pom.xml index 208bc3a..908a601 100644 --- a/idempotent-nats/pom.xml +++ b/idempotent-nats/pom.xml @@ -5,7 +5,7 @@ io.github.arun0009 idempotent - 2.0.1 + 2.1.0 idempotent-nats diff --git a/idempotent-nats/src/main/java/io/github/arun0009/idempotent/nats/NatsIdempotentStore.java b/idempotent-nats/src/main/java/io/github/arun0009/idempotent/nats/NatsIdempotentStore.java index 68cf7fb..592864f 100644 --- a/idempotent-nats/src/main/java/io/github/arun0009/idempotent/nats/NatsIdempotentStore.java +++ b/idempotent-nats/src/main/java/io/github/arun0009/idempotent/nats/NatsIdempotentStore.java @@ -11,7 +11,6 @@ import tools.jackson.databind.json.JsonMapper; import java.io.IOException; -import java.time.Duration; import java.time.Instant; import java.util.Base64; @@ -28,8 +27,9 @@ class NatsIdempotentStore implements IdempotentStore { } private static MessageTtl fromExpirationTimeInMs(long value) { - int ttl = (int) Duration.ofMillis(value - Instant.now().toEpochMilli()).toSeconds(); - return MessageTtl.seconds(ttl + 1); + long ttlMillis = value - Instant.now().toEpochMilli(); + int ttlSeconds = (int) Math.max(1, (ttlMillis + 999) / 1000); // Round up, min 1 second + return MessageTtl.seconds(ttlSeconds); } /** @@ -60,7 +60,14 @@ public Value getValue(IdempotentKey idemKey, Class returnType) { if (entry == null) return null; Wrappers.Value wrapperValue = mapper.readValue(entry.getValue(), Wrappers.Value.class); - return wrapperValue.value(); + Value value = wrapperValue.value(); + + // Check expiration using JVM time + if (value.isExpired()) { + return null; + } + + return value; } catch (IOException | JetStreamApiException e) { log.error("Error reading value from nats store", e); return null; @@ -76,24 +83,16 @@ public void store(IdempotentKey idemKey, Value value) { byte[] content = mapper.writeValueAsBytes(new Wrappers.Value(value)); MessageTtl messageTtl = fromExpirationTimeInMs(value.expirationTimeInMilliSeconds()); - - createOrUpdate(key, content, messageTtl); - } catch (IOException | JetStreamApiException e) { - throw new NatsIdempotentExceptions("Error storing value in nats", e); - } - } - - private void createOrUpdate(String key, byte[] value, MessageTtl ttl) throws JetStreamApiException, IOException { - try { - kv.create(key, value, ttl); + // Try atomic create - will fail with 10071 if key already exists + kv.create(key, content, messageTtl); } catch (JetStreamApiException e) { - // Wrong last sequence, the key already exists. if (e.getApiErrorCode() == 10071) { - kv.put(key, value); - log.atTrace().log("Updated key {}", key); + // key already exists - silent fail (race condition) return; } - throw e; + throw new NatsIdempotentExceptions("Error storing value in nats", e); + } catch (IOException e) { + throw new NatsIdempotentExceptions("Error storing value in nats", e); } } @@ -112,12 +111,18 @@ public void remove(IdempotentKey idemKey) { public void update(IdempotentKey idemKey, Value value) { try { log.atDebug().log("Updating key {} with status {}", idemKey, value.status()); - log.atTrace().log(value::toString); var key = encodeIfNotValid(idemKey); + KeyValueEntry entry = kv.get(key); + if (entry == null) return; // key doesn't exist, silent fail byte[] content = mapper.writeValueAsBytes(new Wrappers.Value(value)); - kv.put(key, content); - } catch (IOException | JetStreamApiException e) { - throw new NatsIdempotentExceptions("Error storing value in nats", e); + kv.update(key, content, entry.getRevision()); // Atomic CAS update + } catch (JetStreamApiException e) { + if (e.getApiErrorCode() == 10071) { + return; // CAS failed, silent fail + } + throw new NatsIdempotentExceptions("Error updating value in nats", e); + } catch (IOException e) { + throw new NatsIdempotentExceptions("Error updating value in nats", e); } } } diff --git a/idempotent-nats/src/main/java/io/github/arun0009/idempotent/nats/NoopNatsAutoConfiguration.java b/idempotent-nats/src/main/java/io/github/arun0009/idempotent/nats/NoopNatsAutoConfiguration.java index eef5c91..dadda2b 100644 --- a/idempotent-nats/src/main/java/io/github/arun0009/idempotent/nats/NoopNatsAutoConfiguration.java +++ b/idempotent-nats/src/main/java/io/github/arun0009/idempotent/nats/NoopNatsAutoConfiguration.java @@ -28,12 +28,16 @@ public Value getValue(IdempotentKey key, Class returnType) { } @Override - public void store(IdempotentKey key, Value value) {} + public void store(IdempotentKey key, Value value) { + // No-op store + } @Override public void remove(IdempotentKey key) {} @Override - public void update(IdempotentKey key, Value value) {} + public void update(IdempotentKey key, Value value) { + // No-op update + } } } diff --git a/idempotent-rds/README.md b/idempotent-rds/README.md new file mode 100644 index 0000000..3e921c5 --- /dev/null +++ b/idempotent-rds/README.md @@ -0,0 +1,83 @@ +# Idempotent Cache with RDS Storage (JDBC) + +To integrate the idempotent cache with RDS (JDBC) storage into your project, add the following dependency to your pom.xml file: + +```xml + + io.github.arun0009 + idempotent-rds + ${idempotent.version} + +``` + +## Overview + +This module provides an idempotent request handling mechanism using a Relational Database (via JDBC/JdbcTemplate) for storage. It requires a database table to be created. + +## Database Schema + +You must create a table for storing idempotent keys. The default table name is `idempotent`. + +**MySQL / PostgreSQL Example:** + +```sql +CREATE TABLE idempotent ( + key_id VARCHAR(255) NOT NULL, + process_name VARCHAR(255) NOT NULL, + status VARCHAR(50), + expiration_time_millis BIGINT, + response TEXT, + PRIMARY KEY (key_id, process_name) +); + +CREATE INDEX idx_expiration_time ON idempotent(expiration_time_millis); +``` + +## Configuration Properties + +### General Properties + +See main [README](../README.md) for general idempotent configuration. + +### RDS Configuration + +* Table Name + + Property: `idempotent.rds.table.name` + Default Value: `idempotent` + Description: The name of the database table to use. + +* Cleanup Schedule + + Property: `idempotent.rds.cleanup.fixedDelay` + Default Value: `60000` (1 minute) + Description: Fixed delay in milliseconds for the cleanup task that removes expired keys. + +* Cleanup Batch Size + + Property: `idempotent.rds.cleanup.batch.size` + Default Value: `1000` + Description: Number of expired keys to delete in each batch to prevent long-running database locks. + +## Distributed Systems Support + +This module is designed to work in a distributed environment with multiple application instances. + +1. **Atomic Operations**: It relies on the database's primary key constraints (`key_id`, `process_name`) to ensure that only one instance can successfully `INSERT` (lock) a specific key. +2. **Concurrency**: If multiple containers try to process the same request simultaneously, the database will reject duplicate inserts with a constraint violation. The application handles this by identifying it as a duplicate request. +3. **Cleanup**: The `RdsCleanupTask` runs on all instances by default. It executes batched `DELETE` statements for expired rows to prevent long-running database locks. This is safe to run concurrently as database transactions ensure consistency (deleting an already deleted row is a no-op). + +## Dependencies + +This module relies on `spring-boot-starter-jdbc`. You must configure your `DataSource` as per standard Spring Boot configuration (e.g., `spring.datasource.url`, etc.). + +## Performance Tuning + +Since this implementation relies on a relational database, performance is critical. Here are some tips to ensure low latency: + +1. **Indexes**: The provided schema uses a composite Primary Key `(key_id, process_name)`. This creates a clustered index (in MySQL/InnoDB) which is the fastest way to look up records. **Do not remove this.** +2. **Connection Pooling**: Use a production-grade connection pool like HikariCP (default in Spring Boot). + * Set `maximum-pool-size` appropriately for your concurrency level. + * Set `minimum-idle` to keep connections warm. +3. **Payload Size**: The `response` column stores the JSON response. If your responses are very large (MBs), retrieving them will add latency. Consider keeping responses concise or using a hybrid approach if payloads are massive. +4. **Database Hardware**: Ensure your RDS instance has sufficient IOPS, as idempotency checks involve frequent reads and writes. diff --git a/idempotent-rds/pom.xml b/idempotent-rds/pom.xml new file mode 100644 index 0000000..6c314a6 --- /dev/null +++ b/idempotent-rds/pom.xml @@ -0,0 +1,102 @@ + + + 4.0.0 + + io.github.arun0009 + idempotent + 2.1.0 + + idempotent-rds + jar + idempotent-rds + idempotent-rds + https://github.com/arun0009/idempotent/tree/main/idempotent-rds + + + 1.21.4 + + + + + + org.testcontainers + junit-jupiter + ${testcontainers.version} + test + + + org.testcontainers + mysql + ${testcontainers.version} + test + + + org.testcontainers + postgresql + ${testcontainers.version} + test + + + + + + + io.github.arun0009 + idempotent-core + + + org.springframework.boot + spring-boot-starter-jdbc + + + org.springframework.boot + spring-boot-starter-test + test + + + + org.testcontainers + junit-jupiter + test + + + org.testcontainers + mysql + test + + + org.testcontainers + postgresql + test + + + org.testcontainers + testcontainers + test + + + + com.mysql + mysql-connector-j + test + + + org.postgresql + postgresql + test + + + + com.h2database + h2 + test + + + io.github.arun0009 + idempotent-core + test-jar + test + + + diff --git a/idempotent-rds/src/main/java/io/github/arun0009/idempotent/rds/RdsCleanupTask.java b/idempotent-rds/src/main/java/io/github/arun0009/idempotent/rds/RdsCleanupTask.java new file mode 100644 index 0000000..b352fc4 --- /dev/null +++ b/idempotent-rds/src/main/java/io/github/arun0009/idempotent/rds/RdsCleanupTask.java @@ -0,0 +1,59 @@ +package io.github.arun0009.idempotent.rds; + +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; +import org.springframework.jdbc.core.JdbcTemplate; +import org.springframework.scheduling.annotation.Scheduled; + +public class RdsCleanupTask { + + private static final Logger log = LoggerFactory.getLogger(RdsCleanupTask.class); + private final JdbcTemplate jdbcTemplate; + private final String tableName; + private final RdsDialect dialect; + private final int batchSize; + + public RdsCleanupTask(JdbcTemplate jdbcTemplate, String tableName, RdsDialect dialect, int batchSize) { + this.jdbcTemplate = jdbcTemplate; + this.tableName = tableName; + this.dialect = dialect; + this.batchSize = batchSize; + } + + @Scheduled(fixedDelayString = "${idempotent.rds.cleanup.fixedDelay:60000}") + public void cleanup() { + long now = System.currentTimeMillis(); + int totalDeleted = 0; + int batchDeleted; + + do { + batchDeleted = deleteBatch(now); + totalDeleted += batchDeleted; + } while (batchDeleted > 0 && batchDeleted == batchSize); + + if (totalDeleted > 0) { + log.debug("Deleted {} expired idempotent keys", totalDeleted); + } + } + + private int deleteBatch(long now) { + String sql; + if (dialect == RdsDialect.MYSQL) { + sql = """ + DELETE FROM %s WHERE expiration_time_millis < ? LIMIT ? + """.formatted(tableName); + return jdbcTemplate.update(sql, now, batchSize); + } else if (dialect == RdsDialect.POSTGRES) { + sql = """ + DELETE FROM %s WHERE ctid IN (SELECT ctid FROM %s WHERE expiration_time_millis < ? LIMIT ?) + """.formatted(tableName, tableName); + return jdbcTemplate.update(sql, now, batchSize); + } else { + // Generic or H2 (H2 in MySQL mode supports LIMIT) + sql = """ + DELETE FROM %s WHERE expiration_time_millis < ? LIMIT ? + """.formatted(tableName); + return jdbcTemplate.update(sql, now, batchSize); + } + } +} diff --git a/idempotent-rds/src/main/java/io/github/arun0009/idempotent/rds/RdsConfig.java b/idempotent-rds/src/main/java/io/github/arun0009/idempotent/rds/RdsConfig.java new file mode 100644 index 0000000..5718699 --- /dev/null +++ b/idempotent-rds/src/main/java/io/github/arun0009/idempotent/rds/RdsConfig.java @@ -0,0 +1,41 @@ +package io.github.arun0009.idempotent.rds; + +import io.github.arun0009.idempotent.core.persistence.IdempotentStore; +import org.springframework.beans.factory.annotation.Value; +import org.springframework.boot.autoconfigure.AutoConfigureAfter; +import org.springframework.boot.autoconfigure.condition.ConditionalOnBean; +import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean; +import org.springframework.context.annotation.Bean; +import org.springframework.context.annotation.Configuration; +import org.springframework.jdbc.core.JdbcTemplate; +import org.springframework.scheduling.annotation.EnableScheduling; + +import javax.sql.DataSource; + +@Configuration +@EnableScheduling +@ConditionalOnBean(DataSource.class) +@AutoConfigureAfter( + name = { + "org.springframework.boot.autoconfigure.jdbc.DataSourceAutoConfiguration", + "org.springframework.boot.autoconfigure.jdbc.JdbcTemplateAutoConfiguration" + }) +public class RdsConfig { + + @Bean + @ConditionalOnMissingBean + public IdempotentStore idempotentStore( + JdbcTemplate jdbcTemplate, @Value("${idempotent.rds.table.name:idempotent}") String tableName) { + return new RdsIdempotentStore(jdbcTemplate, tableName); + } + + @Bean + @ConditionalOnMissingBean + public RdsCleanupTask rdsCleanupTask( + JdbcTemplate jdbcTemplate, + @Value("${idempotent.rds.table.name:idempotent}") String tableName, + @Value("${idempotent.rds.cleanup.batch.size:1000}") int batchSize) { + RdsDialect dialect = RdsDialect.detect(jdbcTemplate); + return new RdsCleanupTask(jdbcTemplate, tableName, dialect, batchSize); + } +} diff --git a/idempotent-rds/src/main/java/io/github/arun0009/idempotent/rds/RdsDialect.java b/idempotent-rds/src/main/java/io/github/arun0009/idempotent/rds/RdsDialect.java new file mode 100644 index 0000000..884559c --- /dev/null +++ b/idempotent-rds/src/main/java/io/github/arun0009/idempotent/rds/RdsDialect.java @@ -0,0 +1,32 @@ +package io.github.arun0009.idempotent.rds; + +import org.springframework.jdbc.core.JdbcTemplate; + +import java.sql.Connection; +import java.sql.DatabaseMetaData; + +public enum RdsDialect { + POSTGRES, + MYSQL, + H2, + GENERIC; + + public static RdsDialect detect(JdbcTemplate jdbcTemplate) { + try { + return jdbcTemplate.execute((Connection conn) -> { + DatabaseMetaData metaData = conn.getMetaData(); + String databaseProductName = metaData.getDatabaseProductName().toLowerCase(); + if (databaseProductName.contains("postgresql")) { + return POSTGRES; + } else if (databaseProductName.contains("mysql") || databaseProductName.contains("mariadb")) { + return MYSQL; + } else if (databaseProductName.contains("h2")) { + return H2; + } + return GENERIC; + }); + } catch (Exception e) { + return GENERIC; + } + } +} diff --git a/idempotent-rds/src/main/java/io/github/arun0009/idempotent/rds/RdsIdempotentStore.java b/idempotent-rds/src/main/java/io/github/arun0009/idempotent/rds/RdsIdempotentStore.java new file mode 100644 index 0000000..b8cb9d3 --- /dev/null +++ b/idempotent-rds/src/main/java/io/github/arun0009/idempotent/rds/RdsIdempotentStore.java @@ -0,0 +1,198 @@ +package io.github.arun0009.idempotent.rds; + +import io.github.arun0009.idempotent.core.exception.IdempotentException; +import io.github.arun0009.idempotent.core.persistence.IdempotentStore; +import org.springframework.dao.DuplicateKeyException; +import org.springframework.dao.EmptyResultDataAccessException; +import org.springframework.jdbc.core.JdbcTemplate; +import org.springframework.jdbc.core.RowMapper; +import tools.jackson.core.JacksonException; +import tools.jackson.databind.json.JsonMapper; + +import java.sql.ResultSet; +import java.sql.SQLException; + +/** + * RDS idempotent store using JdbcTemplate with atomic race condition protection. + */ +public class RdsIdempotentStore implements IdempotentStore { + + private final JdbcTemplate jdbcTemplate; + private final String tableName; + private final JsonMapper jsonMapper = JsonMapper.shared(); + private final RdsDialect dialect; + + public RdsIdempotentStore(JdbcTemplate jdbcTemplate, String tableName) { + this.jdbcTemplate = jdbcTemplate; + this.tableName = tableName; + this.dialect = RdsDialect.detect(jdbcTemplate); + } + + @Override + public Value getValue(IdempotentKey key, Class returnType) { + String sql = """ + SELECT status, expiration_time_millis, response FROM %s WHERE key_id = ? AND process_name = ? + """.formatted(tableName); + try { + Value value = jdbcTemplate.queryForObject( + sql, + new RowMapper() { + @Override + public Value mapRow(ResultSet rs, int rowNum) throws SQLException { + try { + String status = rs.getString("status"); + Long expirationTime = rs.getLong("expiration_time_millis"); + if (rs.wasNull()) { + expirationTime = null; + } + String responseJson = rs.getString("response"); + Object response = null; + if (responseJson != null) { + response = jsonMapper.readValue(responseJson, returnType); + } + Value result = new Value(status, expirationTime, response); + // Check expiration using JVM time + if (result.isExpired()) { + return null; // Treat expired as not found + } + return result; + } catch (JacksonException e) { + throw new IdempotentException("Error deserializing response", e); + } + } + }, + key.key(), + key.processName()); + + // Additional expiration check + if (value != null && value.isExpired()) { + return null; + } + return value; + } catch (EmptyResultDataAccessException e) { + return null; + } + } + + @Override + public void store(IdempotentKey key, Value value) { + try { + String responseJson = value.response() != null ? jsonMapper.writeValueAsString(value.response()) : null; + long now = System.currentTimeMillis(); + + if (dialect == RdsDialect.POSTGRES) { + storePostgres(key, value, responseJson, now); + } else if (dialect == RdsDialect.MYSQL) { + storeMySQL(key, value, responseJson, now); + } else { + storeGeneric(key, value, responseJson, now); + } + + } catch (JacksonException e) { + throw new IdempotentException("Error serializing response", e); + } + } + + private void storePostgres(IdempotentKey key, Value value, String responseJson, long now) { + // Postgres: use ON CONFLICT with conditional WHERE clause + String sql = """ + INSERT INTO %s (key_id, process_name, status, expiration_time_millis, response) + VALUES (?, ?, ?, ?, ?) + ON CONFLICT (key_id, process_name) + DO UPDATE SET + status = EXCLUDED.status, + expiration_time_millis = EXCLUDED.expiration_time_millis, + response = EXCLUDED.response + WHERE %s.expiration_time_millis < ? + """.formatted(tableName, tableName); + jdbcTemplate.update( + sql, + key.key(), + key.processName(), + value.status(), + value.expirationTimeInMilliSeconds(), + responseJson, + now); + } + + private void storeMySQL(IdempotentKey key, Value value, String responseJson, long now) { + // MySQL: use ON DUPLICATE KEY UPDATE with IF() condition + String insertSql = """ + INSERT INTO %s (key_id, process_name, status, expiration_time_millis, response) VALUES (?, ?, ?, ?, ?) + ON DUPLICATE KEY UPDATE + status = IF(expiration_time_millis < ?, VALUES(status), status), + expiration_time_millis = IF(expiration_time_millis < ?, VALUES(expiration_time_millis), expiration_time_millis), + response = IF(expiration_time_millis < ?, VALUES(response), response) + """.formatted(tableName); + jdbcTemplate.update( + insertSql, + key.key(), + key.processName(), + value.status(), + value.expirationTimeInMilliSeconds(), + responseJson, + now, + now, + now); + } + + private void storeGeneric(IdempotentKey key, Value value, String response, long now) { + // Generic H2: INSERT first, if duplicate then try UPDATE with WHERE clause + try { + String insertSQL = """ + INSERT INTO %s (key_id, process_name, status, expiration_time_millis, response) VALUES (?, ?, ?, ?, ?) + """.formatted(tableName); + jdbcTemplate.update( + insertSQL, + key.key(), + key.processName(), + value.status(), + value.expirationTimeInMilliSeconds(), + response); + } catch (DuplicateKeyException e) { + // Key exists, try to update only if expired + String updateSql = """ + UPDATE %s SET status = ?, expiration_time_millis = ?, response = ? + WHERE key_id = ? AND process_name = ? AND expiration_time_millis < ? + """.formatted(tableName); + jdbcTemplate.update( + updateSql, + value.status(), + value.expirationTimeInMilliSeconds(), + response, + key.key(), + key.processName(), + now); + // IF UPDATE affects 0 rows, key exists but not expired - silent fail + } + } + + @Override + public void remove(IdempotentKey key) { + String sql = """ + DELETE FROM %s WHERE key_id = ? AND process_name = ? + """.formatted(tableName); + jdbcTemplate.update(sql, key.key(), key.processName()); + } + + @Override + public void update(IdempotentKey key, Value value) { + try { + String responseJson = value.response() != null ? jsonMapper.writeValueAsString(value.response()) : null; + // Simple UPDATE - trust the caller's contract that this is for INPROGRESS keys + String updateSql = """ + UPDATE %s SET status = ?, expiration_time_millis = ?, response = ? + WHERE key_id = ? AND process_name = ? + """.formatted(tableName); + jdbcTemplate.update( + updateSql, + value.status(), + value.expirationTimeInMilliSeconds(), + responseJson, + key.key(), + key.processName()); + } catch (JacksonException e) { + throw new IdempotentException("Error serializing response", e); + } + } +} diff --git a/idempotent-rds/src/main/resources/META-INF/spring/org.springframework.boot.autoconfigure.AutoConfiguration.imports b/idempotent-rds/src/main/resources/META-INF/spring/org.springframework.boot.autoconfigure.AutoConfiguration.imports new file mode 100644 index 0000000..36345c8 --- /dev/null +++ b/idempotent-rds/src/main/resources/META-INF/spring/org.springframework.boot.autoconfigure.AutoConfiguration.imports @@ -0,0 +1 @@ +io.github.arun0009.idempotent.rds.RdsConfig diff --git a/idempotent-rds/src/test/java/io/github/arun0009/idempotent/rds/RdsIdempotentStoreH2Test.java b/idempotent-rds/src/test/java/io/github/arun0009/idempotent/rds/RdsIdempotentStoreH2Test.java new file mode 100644 index 0000000..1de33e7 --- /dev/null +++ b/idempotent-rds/src/test/java/io/github/arun0009/idempotent/rds/RdsIdempotentStoreH2Test.java @@ -0,0 +1,126 @@ +package io.github.arun0009.idempotent.rds; + +import io.github.arun0009.idempotent.core.persistence.IdempotentStore; +import io.github.arun0009.idempotent.core.persistence.IdempotentStore.IdempotentKey; +import io.github.arun0009.idempotent.core.persistence.IdempotentStore.Value; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.extension.ExtendWith; +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.boot.jdbc.DataSourceBuilder; +import org.springframework.context.annotation.Bean; +import org.springframework.context.annotation.Configuration; +import org.springframework.jdbc.core.JdbcTemplate; +import org.springframework.test.context.ContextConfiguration; +import org.springframework.test.context.TestPropertySource; +import org.springframework.test.context.junit.jupiter.SpringExtension; + +import javax.sql.DataSource; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertNotNull; +import static org.junit.jupiter.api.Assertions.assertNull; + +/** + * Fast H2-based tests for development. + * These tests use H2 in MySQL mode for quick feedback during development. + * For production accuracy, use MySQLTest and PostgreSQLTest. + */ +@ExtendWith(SpringExtension.class) +@ContextConfiguration(classes = RdsIdempotentStoreH2Test.H2TestConfig.class) +@TestPropertySource(properties = {"idempotent.rds.table.name=idempotent"}) +public class RdsIdempotentStoreH2Test { + + @Autowired + private IdempotentStore idempotentStore; + + @Autowired + private RdsCleanupTask rdsCleanupTask; + + @Autowired + private JdbcTemplate jdbcTemplate; + + @Configuration + static class H2TestConfig { + @Bean + public DataSource dataSource() { + return DataSourceBuilder.create() + .url("jdbc:h2:mem:testdb;DB_CLOSE_DELAY=-1;MODE=MySQL") + .driverClassName("org.h2.Driver") + .username("sa") + .password("") + .build(); + } + + @Bean + public JdbcTemplate jdbcTemplate(DataSource dataSource) { + return new JdbcTemplate(dataSource); + } + + @Bean + public IdempotentStore idempotentStore(JdbcTemplate jdbcTemplate) { + return new RdsIdempotentStore(jdbcTemplate, "idempotent"); + } + + @Bean + public RdsCleanupTask rdsCleanupTask(JdbcTemplate jdbcTemplate) { + RdsDialect dialect = RdsDialect.detect(jdbcTemplate); + int batchSize = 1000; + return new RdsCleanupTask(jdbcTemplate, "idempotent", dialect, batchSize); + } + } + + @BeforeEach + public void setUp() { + // Create table for H2 + jdbcTemplate.update(""" + CREATE TABLE IF NOT EXISTS idempotent ( + key_id VARCHAR(255) NOT NULL, + process_name VARCHAR(255) NOT NULL, + status VARCHAR(50), + expiration_time_millis BIGINT, + response TEXT, + PRIMARY KEY (key_id, process_name) + ) + """); + + jdbcTemplate.update("CREATE INDEX IF NOT EXISTS idx_expiration_time ON idempotent(expiration_time_millis)"); + jdbcTemplate.update("DELETE FROM idempotent"); + } + + @Test + public void testStoreAndGet() { + IdempotentKey key = new IdempotentKey("test-key", "test-process"); + Value value = new Value("COMPLETED", System.currentTimeMillis() + 5000, "success"); + + idempotentStore.store(key, value); + + Value retrieved = idempotentStore.getValue(key, String.class); + assertNotNull(retrieved); + assertEquals("COMPLETED", retrieved.status()); + assertEquals("success", retrieved.response()); + } + + @Test + public void testDialectDetection() { + // Verify H2 dialect is detected correctly + RdsDialect dialect = RdsDialect.detect(jdbcTemplate); + assertEquals(RdsDialect.H2, dialect); + } + + @Test + public void testRaceConditionProtection() { + IdempotentKey key = new IdempotentKey("race-key", "race-process"); + Value value1 = new Value("INPROGRESS", System.currentTimeMillis() + 10000, null); + idempotentStore.store(key, value1); + + Value value2 = new Value("COMPLETED", System.currentTimeMillis() + 20000, "overwritten"); + // This should NOT overwrite value1 due to race condition protection + idempotentStore.store(key, value2); + + Value retrieved = idempotentStore.getValue(key, String.class); + assertNotNull(retrieved); + assertEquals("INPROGRESS", retrieved.status()); // Should still be original + assertNull(retrieved.response()); // Original had null response + } +} diff --git a/idempotent-rds/src/test/java/io/github/arun0009/idempotent/rds/RdsIdempotentStoreMySQLTest.java b/idempotent-rds/src/test/java/io/github/arun0009/idempotent/rds/RdsIdempotentStoreMySQLTest.java new file mode 100644 index 0000000..5ca368d --- /dev/null +++ b/idempotent-rds/src/test/java/io/github/arun0009/idempotent/rds/RdsIdempotentStoreMySQLTest.java @@ -0,0 +1,166 @@ +package io.github.arun0009.idempotent.rds; + +import io.github.arun0009.idempotent.core.persistence.IdempotentStore; +import io.github.arun0009.idempotent.core.persistence.IdempotentStore.IdempotentKey; +import io.github.arun0009.idempotent.core.persistence.IdempotentStore.Value; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.extension.ExtendWith; +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.jdbc.core.JdbcTemplate; +import org.springframework.test.context.ContextConfiguration; +import org.springframework.test.context.TestPropertySource; +import org.springframework.test.context.junit.jupiter.SpringExtension; + +import java.util.List; +import java.util.Map; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertNotNull; +import static org.junit.jupiter.api.Assertions.assertNull; + +@ExtendWith(SpringExtension.class) +@ContextConfiguration(classes = RdsMySQLTestConfig.class) +@TestPropertySource(properties = {"idempotent.rds.table.name=idempotent"}) +public class RdsIdempotentStoreMySQLTest { + + @Autowired + private IdempotentStore idempotentStore; + + @Autowired + private RdsCleanupTask rdsCleanupTask; + + @Autowired + private JdbcTemplate jdbcTemplate; + + @BeforeEach + public void setUp() { + // Create the table for MySQL + jdbcTemplate.update(""" + CREATE TABLE IF NOT EXISTS idempotent ( + key_id VARCHAR(255) NOT NULL, + process_name VARCHAR(255) NOT NULL, + status VARCHAR(50), + expiration_time_millis BIGINT, + response TEXT, + PRIMARY KEY (key_id, process_name) + ) + """); + + // MySQL doesn't support CREATE INDEX IF NOT EXISTS, so we check first + try { + jdbcTemplate.update("CREATE INDEX idx_expiration_time ON idempotent(expiration_time_millis)"); + } catch (Exception e) { + // Index already exists, ignore + } + + // Clean up the database before each test + jdbcTemplate.update("DELETE FROM idempotent"); + } + + @Test + public void testStoreAndGet() { + IdempotentKey key = new IdempotentKey("test-key", "test-process"); + Value value = new Value("COMPLETED", System.currentTimeMillis() + 5000, Map.of("result", "success")); + + idempotentStore.store(key, value); + + Value retrieved = idempotentStore.getValue(key, Map.class); + assertNotNull(retrieved); + assertEquals("COMPLETED", retrieved.status()); + assertEquals("success", ((Map) retrieved.response()).get("result")); + } + + @Test + public void testUpdate() { + IdempotentKey key = new IdempotentKey("test-key-update", "test-process"); + Value value = new Value("INPROGRESS", System.currentTimeMillis() + 10000, null); + + idempotentStore.store(key, value); + + Value newValue = new Value("COMPLETED", System.currentTimeMillis() + 20000, Map.of("result", "updated")); + idempotentStore.update(key, newValue); + + Value retrieved = idempotentStore.getValue(key, Map.class); + assertNotNull(retrieved); + assertEquals("COMPLETED", retrieved.status()); // INPROGRESS entries can be updated + if (retrieved.response() != null) { + assertEquals("updated", ((Map) retrieved.response()).get("result")); + } + } + + @Test + public void testDuplicateStoreDoesNotOverwrite() { + IdempotentKey key = new IdempotentKey("dup-key", "dup-process"); + Value value1 = new Value("INPROGRESS", System.currentTimeMillis() + 10000, null); + idempotentStore.store(key, value1); + + Value value2 = new Value("COMPLETED", System.currentTimeMillis() + 20000, Map.of("data", "overwritten")); + + // With race condition protection, this should NOT overwrite value1 + idempotentStore.store(key, value2); + + Value retrieved = idempotentStore.getValue(key, Map.class); + assertNotNull(retrieved); + assertEquals("INPROGRESS", retrieved.status()); // Should still be the original value + assertNull(retrieved.response()); // Original value had null response + } + + @Test + public void testCleanup() { + IdempotentKey key1 = new IdempotentKey("test-key-cleanup-1", "test-process"); + // Expired + Value value1 = new Value("COMPLETED", System.currentTimeMillis() - 10000, Map.of("data", "expired")); + + IdempotentKey key2 = new IdempotentKey("test-key-cleanup-2", "test-process"); + // Not expired + Value value2 = new Value("COMPLETED", System.currentTimeMillis() + 10000, Map.of("data", "valid")); + + idempotentStore.store(key1, value1); + idempotentStore.store(key2, value2); + + rdsCleanupTask.cleanup(); + + assertNull(idempotentStore.getValue(key1, Map.class)); + assertNotNull(idempotentStore.getValue(key2, Map.class)); + } + + @Test + public void testComplexPojoSerialization() { + IdempotentKey key = new IdempotentKey("complex-key", "complex-process"); + Map complexData = Map.of( + "id", + 123, + "status", + "active", + "tags", + List.of("tag1", "tag2"), + "metadata", + Map.of("source", "api", "version", "1.0")); + + Value value = new Value("COMPLETED", System.currentTimeMillis() + 5000, complexData); + + idempotentStore.store(key, value); + + Value retrieved = idempotentStore.getValue(key, Map.class); + assertNotNull(retrieved); + assertEquals("COMPLETED", retrieved.status()); + Map response = (Map) retrieved.response(); + assertEquals(123, response.get("id")); + assertEquals("active", response.get("status")); + assertEquals(List.of("tag1", "tag2"), response.get("tags")); + } + + @Test + public void testMySQLDialectDetection() { + // Verify that MySQL dialect is detected correctly with real MySQL + RdsDialect dialect = RdsDialect.detect(jdbcTemplate); + assertEquals(RdsDialect.MYSQL, dialect); + + // Verify MySQL syntax works by testing MySQL-specific features + jdbcTemplate.update( + "CREATE TABLE IF NOT EXISTS test_mysql_syntax (id INT AUTO_INCREMENT PRIMARY KEY, name TEXT)"); + jdbcTemplate.update("INSERT INTO test_mysql_syntax (name) VALUES ('test')"); + jdbcTemplate.update("DROP TABLE test_mysql_syntax"); + } +} diff --git a/idempotent-rds/src/test/java/io/github/arun0009/idempotent/rds/RdsIdempotentStorePostgreSQLTest.java b/idempotent-rds/src/test/java/io/github/arun0009/idempotent/rds/RdsIdempotentStorePostgreSQLTest.java new file mode 100644 index 0000000..fee6914 --- /dev/null +++ b/idempotent-rds/src/test/java/io/github/arun0009/idempotent/rds/RdsIdempotentStorePostgreSQLTest.java @@ -0,0 +1,158 @@ +package io.github.arun0009.idempotent.rds; + +import io.github.arun0009.idempotent.core.persistence.IdempotentStore; +import io.github.arun0009.idempotent.core.persistence.IdempotentStore.IdempotentKey; +import io.github.arun0009.idempotent.core.persistence.IdempotentStore.Value; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.extension.ExtendWith; +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.jdbc.core.JdbcTemplate; +import org.springframework.test.context.ContextConfiguration; +import org.springframework.test.context.TestPropertySource; +import org.springframework.test.context.junit.jupiter.SpringExtension; + +import java.util.List; +import java.util.Map; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertNotNull; +import static org.junit.jupiter.api.Assertions.assertNull; + +@ExtendWith(SpringExtension.class) +@ContextConfiguration(classes = RdsPostgreSQLTestConfig.class) +@TestPropertySource(properties = {"idempotent.rds.table.name=idempotent"}) +public class RdsIdempotentStorePostgreSQLTest { + + @Autowired + private IdempotentStore idempotentStore; + + @Autowired + private RdsCleanupTask rdsCleanupTask; + + @Autowired + private JdbcTemplate jdbcTemplate; + + @BeforeEach + public void setUp() { + // Create the table for PostgreSQL + jdbcTemplate.update(""" + CREATE TABLE IF NOT EXISTS idempotent ( + key_id VARCHAR(255) NOT NULL, + process_name VARCHAR(255) NOT NULL, + status VARCHAR(50), + expiration_time_millis BIGINT, + response TEXT, + PRIMARY KEY (key_id, process_name) + ) + """); + + jdbcTemplate.update("CREATE INDEX IF NOT EXISTS idx_expiration_time ON idempotent(expiration_time_millis)"); + + // Clean up the database before each test + jdbcTemplate.update("DELETE FROM idempotent"); + } + + @Test + public void testStoreAndGet() { + IdempotentKey key = new IdempotentKey("test-key", "test-process"); + Value value = new Value("COMPLETED", System.currentTimeMillis() + 5000, Map.of("result", "success")); + + idempotentStore.store(key, value); + + Value retrieved = idempotentStore.getValue(key, Map.class); + assertNotNull(retrieved); + assertEquals("COMPLETED", retrieved.status()); + assertEquals("success", ((Map) retrieved.response()).get("result")); + } + + @Test + public void testUpdate() { + IdempotentKey key = new IdempotentKey("test-key-update", "test-process"); + Value value = new Value("INPROGRESS", System.currentTimeMillis() + 10000, null); + + idempotentStore.store(key, value); + + Value newValue = new Value("COMPLETED", System.currentTimeMillis() + 20000, Map.of("result", "updated")); + idempotentStore.update(key, newValue); + + Value retrieved = idempotentStore.getValue(key, Map.class); + assertNotNull(retrieved); + assertEquals("COMPLETED", retrieved.status()); + assertEquals("updated", ((Map) retrieved.response()).get("result")); + } + + @Test + public void testDuplicateStoreDoesNotOverwrite() { + IdempotentKey key = new IdempotentKey("dup-key", "dup-process"); + Value value1 = new Value("INPROGRESS", System.currentTimeMillis() + 10000, null); + idempotentStore.store(key, value1); + + Value value2 = new Value("COMPLETED", System.currentTimeMillis() + 20000, Map.of("data", "overwritten")); + + // With race condition protection, this should NOT overwrite value1 + idempotentStore.store(key, value2); + + Value retrieved = idempotentStore.getValue(key, Map.class); + assertNotNull(retrieved); + assertEquals("INPROGRESS", retrieved.status()); // Should still be the original value + assertNull(retrieved.response()); // Original value had null response + } + + @Test + public void testCleanup() { + IdempotentKey key1 = new IdempotentKey("test-key-cleanup-1", "test-process"); + // Expired + Value value1 = new Value("COMPLETED", System.currentTimeMillis() - 10000, Map.of("data", "expired")); + + IdempotentKey key2 = new IdempotentKey("test-key-cleanup-2", "test-process"); + // Not expired + Value value2 = new Value("COMPLETED", System.currentTimeMillis() + 10000, Map.of("data", "valid")); + + idempotentStore.store(key1, value1); + idempotentStore.store(key2, value2); + + rdsCleanupTask.cleanup(); + + assertNull(idempotentStore.getValue(key1, Map.class)); + assertNotNull(idempotentStore.getValue(key2, Map.class)); + } + + @Test + public void testComplexPojoSerialization() { + IdempotentKey key = new IdempotentKey("complex-key", "complex-process"); + Map complexData = Map.of( + "id", + 123, + "status", + "active", + "tags", + List.of("tag1", "tag2"), + "metadata", + Map.of("source", "api", "version", "1.0")); + + Value value = new Value("COMPLETED", System.currentTimeMillis() + 5000, complexData); + + idempotentStore.store(key, value); + + Value retrieved = idempotentStore.getValue(key, Map.class); + assertNotNull(retrieved); + assertEquals("COMPLETED", retrieved.status()); + Map response = (Map) retrieved.response(); + assertEquals(123, response.get("id")); + assertEquals("active", response.get("status")); + assertEquals(List.of("tag1", "tag2"), response.get("tags")); + } + + @Test + public void testPostgreSQLDialectDetection() { + // Verify that PostgreSQL dialect is detected correctly with real PostgreSQL + RdsDialect dialect = RdsDialect.detect(jdbcTemplate); + assertEquals(RdsDialect.POSTGRES, dialect); + + // Verify PostgreSQL syntax works (SERIAL, TEXT types) + jdbcTemplate.update("CREATE TABLE IF NOT EXISTS test_pg_syntax (id SERIAL PRIMARY KEY, name TEXT)"); + jdbcTemplate.update("INSERT INTO test_pg_syntax (name) VALUES ('test')"); + jdbcTemplate.update("DROP TABLE test_pg_syntax"); + } +} diff --git a/idempotent-rds/src/test/java/io/github/arun0009/idempotent/rds/RdsMySQLTestConfig.java b/idempotent-rds/src/test/java/io/github/arun0009/idempotent/rds/RdsMySQLTestConfig.java new file mode 100644 index 0000000..84c5542 --- /dev/null +++ b/idempotent-rds/src/test/java/io/github/arun0009/idempotent/rds/RdsMySQLTestConfig.java @@ -0,0 +1,66 @@ +package io.github.arun0009.idempotent.rds; + +import io.github.arun0009.idempotent.core.persistence.IdempotentStore; +import org.springframework.boot.jdbc.DataSourceBuilder; +import org.springframework.context.annotation.Bean; +import org.springframework.context.annotation.Configuration; +import org.springframework.jdbc.core.JdbcTemplate; +import org.springframework.test.context.DynamicPropertyRegistry; +import org.springframework.test.context.DynamicPropertySource; +import org.testcontainers.containers.MySQLContainer; +import org.testcontainers.junit.jupiter.Container; +import org.testcontainers.junit.jupiter.Testcontainers; +import org.testcontainers.utility.DockerImageName; + +import javax.sql.DataSource; + +@Testcontainers +@Configuration +public class RdsMySQLTestConfig { + + @Container + static MySQLContainer mysql = new MySQLContainer<>(DockerImageName.parse("mysql:8.0")) + .withDatabaseName("testdb") + .withUsername("test") + .withPassword("test") + .withReuse(true); + + static { + mysql.start(); + } + + @Bean + public DataSource dataSource() { + return DataSourceBuilder.create() + .url(mysql.getJdbcUrl()) + .driverClassName(mysql.getDriverClassName()) + .username(mysql.getUsername()) + .password(mysql.getPassword()) + .build(); + } + + @Bean + public JdbcTemplate jdbcTemplate(DataSource dataSource) { + return new JdbcTemplate(dataSource); + } + + @Bean + public IdempotentStore idempotentStore(JdbcTemplate jdbcTemplate) { + return new RdsIdempotentStore(jdbcTemplate, "idempotent"); + } + + @Bean + public RdsCleanupTask rdsCleanupTask(JdbcTemplate jdbcTemplate) { + RdsDialect dialect = RdsDialect.detect(jdbcTemplate); + int batchSize = 1000; // Default batch size for tests + return new RdsCleanupTask(jdbcTemplate, "idempotent", dialect, batchSize); + } + + @DynamicPropertySource + static void mysqlProperties(DynamicPropertyRegistry registry) { + registry.add("spring.datasource.url", mysql::getJdbcUrl); + registry.add("spring.datasource.username", mysql::getUsername); + registry.add("spring.datasource.password", mysql::getPassword); + registry.add("spring.datasource.driver-class-name", mysql::getDriverClassName); + } +} diff --git a/idempotent-rds/src/test/java/io/github/arun0009/idempotent/rds/RdsPostgreSQLTestConfig.java b/idempotent-rds/src/test/java/io/github/arun0009/idempotent/rds/RdsPostgreSQLTestConfig.java new file mode 100644 index 0000000..c602a40 --- /dev/null +++ b/idempotent-rds/src/test/java/io/github/arun0009/idempotent/rds/RdsPostgreSQLTestConfig.java @@ -0,0 +1,66 @@ +package io.github.arun0009.idempotent.rds; + +import io.github.arun0009.idempotent.core.persistence.IdempotentStore; +import org.springframework.boot.jdbc.DataSourceBuilder; +import org.springframework.context.annotation.Bean; +import org.springframework.context.annotation.Configuration; +import org.springframework.jdbc.core.JdbcTemplate; +import org.springframework.test.context.DynamicPropertyRegistry; +import org.springframework.test.context.DynamicPropertySource; +import org.testcontainers.containers.PostgreSQLContainer; +import org.testcontainers.junit.jupiter.Container; +import org.testcontainers.junit.jupiter.Testcontainers; +import org.testcontainers.utility.DockerImageName; + +import javax.sql.DataSource; + +@Testcontainers +@Configuration +public class RdsPostgreSQLTestConfig { + + @Container + static PostgreSQLContainer postgres = new PostgreSQLContainer<>(DockerImageName.parse("postgres:15-alpine")) + .withDatabaseName("testdb") + .withUsername("test") + .withPassword("test") + .withReuse(true); + + static { + postgres.start(); + } + + @Bean + public DataSource dataSource() { + return DataSourceBuilder.create() + .url(postgres.getJdbcUrl()) + .driverClassName(postgres.getDriverClassName()) + .username(postgres.getUsername()) + .password(postgres.getPassword()) + .build(); + } + + @Bean + public JdbcTemplate jdbcTemplate(DataSource dataSource) { + return new JdbcTemplate(dataSource); + } + + @Bean + public IdempotentStore idempotentStore(JdbcTemplate jdbcTemplate) { + return new RdsIdempotentStore(jdbcTemplate, "idempotent"); + } + + @Bean + public RdsCleanupTask rdsCleanupTask(JdbcTemplate jdbcTemplate) { + RdsDialect dialect = RdsDialect.detect(jdbcTemplate); + int batchSize = 1000; // Default batch size for tests + return new RdsCleanupTask(jdbcTemplate, "idempotent", dialect, batchSize); + } + + @DynamicPropertySource + static void postgresProperties(DynamicPropertyRegistry registry) { + registry.add("spring.datasource.url", postgres::getJdbcUrl); + registry.add("spring.datasource.username", postgres::getUsername); + registry.add("spring.datasource.password", postgres::getPassword); + registry.add("spring.datasource.driver-class-name", postgres::getDriverClassName); + } +} diff --git a/idempotent-redis/pom.xml b/idempotent-redis/pom.xml index 5fa36d9..897e110 100644 --- a/idempotent-redis/pom.xml +++ b/idempotent-redis/pom.xml @@ -5,7 +5,7 @@ io.github.arun0009 idempotent - 2.0.1 + 2.1.0 idempotent-redis jar diff --git a/idempotent-redis/src/main/java/io/github/arun0009/idempotent/redis/RedisIdempotentStore.java b/idempotent-redis/src/main/java/io/github/arun0009/idempotent/redis/RedisIdempotentStore.java index 361fa38..6219e95 100644 --- a/idempotent-redis/src/main/java/io/github/arun0009/idempotent/redis/RedisIdempotentStore.java +++ b/idempotent-redis/src/main/java/io/github/arun0009/idempotent/redis/RedisIdempotentStore.java @@ -3,15 +3,69 @@ import io.github.arun0009.idempotent.core.persistence.IdempotentStore; import org.springframework.data.redis.core.RedisTemplate; -import java.util.concurrent.TimeUnit; +import java.util.Collections; /** - * Redis idempotent store + * Redis idempotent store with atomic operations using Lua scripts */ public class RedisIdempotentStore implements IdempotentStore { private final RedisTemplate redisTemplate; + private static final String STORE_SCRIPT = """ + local existing = redis.call('GET', KEYS[1]) + local now = tonumber(ARGV[3]) + if not now then + now = 0 + end + + if not existing then + -- Key doesn't exist, safe to store it + local ttl = tonumber(ARGV[2]) + if ttl and type(ttl) == "number" and ttl > 0 and ttl ~= -1 then + redis.call('SET', KEYS[1], ARGV[1], 'PX', ttl) + else + redis.call('SET', KEYS[1], ARGV[1]) + end + return '"1"' + end + + -- Key exists check if expired + local success, val = pcall(cjson.decode, existing) + if success and val and val.expirationTimeInMilliSeconds and val.expirationTimeInMilliSeconds < now then + -- Expired, safe to overwrite + local ttl = tonumber(ARGV[2]) + if ttl and type(ttl) == "number" and ttl > 0 and ttl ~= -1 then + redis.call('SET', KEYS[1], ARGV[1], 'PX', ttl) + else + redis.call('SET', KEYS[1], ARGV[1]) + end + return '"1"' + end + -- Key exists and not expired, dont overwrite + return '"0"'"""; + + private static final String UPDATE_SCRIPT = """ + local existing = redis.call('GET', KEYS[1]) + if not existing then + -- Key doesn't exist, nothing to update + return '"0"' + end + + local success, val = pcall(cjson.decode, existing) + if success and val and val.status == 'INPROGRESS' then + -- Only update if status is INPROGRESS + local ttl = tonumber(ARGV[2]) + if ttl and type(ttl) == "number" and ttl > 0 and ttl ~= -1 then + redis.call('SET', KEYS[1], ARGV[1], 'PX', ttl) + else + redis.call('SET', KEYS[1], ARGV[1]) + end + return '"1"' + end + -- Key exists and not in progress, dont update + return '"0"'"""; + /** * Instantiates a new Redis idempotent store. * @@ -21,6 +75,19 @@ public RedisIdempotentStore(RedisTemplate redisTemplate) { this.redisTemplate = redisTemplate; } + private void executeRedisScript(String script, IdempotentKey key, Value value) { + long now = System.currentTimeMillis(); + long ttl = value.expirationTimeInMilliSeconds() != null ? value.expirationTimeInMilliSeconds() - now : -1; + + redisTemplate.execute( + new org.springframework.data.redis.core.script.DefaultRedisScript<>(script, String.class), + Collections.singletonList(key), + value, + String.valueOf(ttl), + String.valueOf(now)); + // Silent fail - race condition handled by Lua script + } + @Override public Value getValue(IdempotentKey key, Class returnType) { return redisTemplate.opsForValue().get(key); @@ -28,10 +95,7 @@ public Value getValue(IdempotentKey key, Class returnType) { @Override public void store(IdempotentKey key, Value value) { - redisTemplate.opsForValue().set(key, value); - if (value.expirationTimeInMilliSeconds() != null) { - redisTemplate.expire(key, value.expirationTimeInMilliSeconds(), TimeUnit.MILLISECONDS); - } + executeRedisScript(STORE_SCRIPT, key, value); } @Override @@ -41,9 +105,6 @@ public void remove(IdempotentKey key) { @Override public void update(IdempotentKey key, Value value) { - redisTemplate.opsForValue().set(key, value); - if (value.expirationTimeInMilliSeconds() != null) { - redisTemplate.expire(key, value.expirationTimeInMilliSeconds(), TimeUnit.MILLISECONDS); - } + executeRedisScript(UPDATE_SCRIPT, key, value); } } diff --git a/pom.xml b/pom.xml index 06a16a8..3437b1e 100644 --- a/pom.xml +++ b/pom.xml @@ -4,7 +4,7 @@ 4.0.0 io.github.arun0009 idempotent - 2.0.1 + 2.1.0 pom Idempotent Make your APIs Idempotent @@ -28,6 +28,7 @@ idempotent-redis idempotent-dynamo idempotent-nats + idempotent-rds @@ -77,6 +78,12 @@ provided true + + org.assertj + assertj-core + 3.27.7 + test + @@ -114,6 +121,24 @@ testcontainers test + + org.testcontainers + junit-jupiter + ${testcontainers.version} + test + + + org.testcontainers + mysql + ${testcontainers.version} + test + + + org.testcontainers + postgresql + ${testcontainers.version} + test + From b4171dddb06f8a018c408a78c40a167a84501de0 Mon Sep 17 00:00:00 2001 From: Arun Gopalpuri Date: Sat, 21 Feb 2026 23:18:19 -0800 Subject: [PATCH 2/6] addressing pr suggestions --- idempotent-rds/pom.xml | 37 +++-------- .../idempotent/rds/RdsCleanupTask.java | 41 ++++++------ .../arun0009/idempotent/rds/RdsConfig.java | 62 ++++++++++++++++--- .../arun0009/idempotent/rds/RdsDialect.java | 9 +++ .../rds/RdsIdempotentProperties.java | 62 +++++++++++++++++++ .../idempotent/rds/RdsIdempotentStore.java | 18 +++--- .../rds/RdsIdempotentStoreH2Test.java | 23 +++---- .../rds/RdsIdempotentStoreMySQLTest.java | 20 +++--- .../rds/RdsIdempotentStorePostgreSQLTest.java | 20 +++--- .../idempotent/rds/RdsMySQLTestConfig.java | 3 +- .../rds/RdsPostgreSQLTestConfig.java | 3 +- 11 files changed, 200 insertions(+), 98 deletions(-) create mode 100644 idempotent-rds/src/main/java/io/github/arun0009/idempotent/rds/RdsIdempotentProperties.java diff --git a/idempotent-rds/pom.xml b/idempotent-rds/pom.xml index 6c314a6..f12d00c 100644 --- a/idempotent-rds/pom.xml +++ b/idempotent-rds/pom.xml @@ -13,33 +13,6 @@ idempotent-rds https://github.com/arun0009/idempotent/tree/main/idempotent-rds - - 1.21.4 - - - - - - org.testcontainers - junit-jupiter - ${testcontainers.version} - test - - - org.testcontainers - mysql - ${testcontainers.version} - test - - - org.testcontainers - postgresql - ${testcontainers.version} - test - - - - io.github.arun0009 @@ -49,6 +22,13 @@ org.springframework.boot spring-boot-starter-jdbc + + org.springframework.boot + spring-boot-configuration-processor + ${spring-boot.version} + provided + true + org.springframework.boot spring-boot-starter-test @@ -58,16 +38,19 @@ org.testcontainers junit-jupiter + ${testcontainers.version} test org.testcontainers mysql + ${testcontainers.version} test org.testcontainers postgresql + ${testcontainers.version} test diff --git a/idempotent-rds/src/main/java/io/github/arun0009/idempotent/rds/RdsCleanupTask.java b/idempotent-rds/src/main/java/io/github/arun0009/idempotent/rds/RdsCleanupTask.java index b352fc4..2600aee 100644 --- a/idempotent-rds/src/main/java/io/github/arun0009/idempotent/rds/RdsCleanupTask.java +++ b/idempotent-rds/src/main/java/io/github/arun0009/idempotent/rds/RdsCleanupTask.java @@ -3,7 +3,6 @@ import org.slf4j.Logger; import org.slf4j.LoggerFactory; import org.springframework.jdbc.core.JdbcTemplate; -import org.springframework.scheduling.annotation.Scheduled; public class RdsCleanupTask { @@ -20,7 +19,6 @@ public RdsCleanupTask(JdbcTemplate jdbcTemplate, String tableName, RdsDialect di this.batchSize = batchSize; } - @Scheduled(fixedDelayString = "${idempotent.rds.cleanup.fixedDelay:60000}") public void cleanup() { long now = System.currentTimeMillis(); int totalDeleted = 0; @@ -37,23 +35,26 @@ public void cleanup() { } private int deleteBatch(long now) { - String sql; - if (dialect == RdsDialect.MYSQL) { - sql = """ - DELETE FROM %s WHERE expiration_time_millis < ? LIMIT ? - """.formatted(tableName); - return jdbcTemplate.update(sql, now, batchSize); - } else if (dialect == RdsDialect.POSTGRES) { - sql = """ - DELETE FROM %s WHERE ctid IN (SELECT ctid FROM %s WHERE expiration_time_millis < ? LIMIT ?) - """.formatted(tableName, tableName); - return jdbcTemplate.update(sql, now, batchSize); - } else { - // Generic or H2 (H2 in MySQL mode supports LIMIT) - sql = """ - DELETE FROM %s WHERE expiration_time_millis < ? LIMIT ? - """.formatted(tableName); - return jdbcTemplate.update(sql, now, batchSize); - } + return switch (dialect) { + case MYSQL -> { + String sql = """ + DELETE FROM %s WHERE expiration_time_millis < ? LIMIT ? + """.formatted(tableName); + yield jdbcTemplate.update(sql, now, batchSize); + } + case POSTGRES -> { + String sql = """ + DELETE FROM %s WHERE ctid IN (SELECT ctid FROM %s WHERE expiration_time_millis < ? LIMIT ?) + """.formatted(tableName, tableName); + yield jdbcTemplate.update(sql, now, batchSize); + } + case H2, GENERIC -> { + // H2 in MySQL mode supports LIMIT + String sql = """ + DELETE FROM %s WHERE expiration_time_millis < ? LIMIT ? + """.formatted(tableName); + yield jdbcTemplate.update(sql, now, batchSize); + } + }; } } diff --git a/idempotent-rds/src/main/java/io/github/arun0009/idempotent/rds/RdsConfig.java b/idempotent-rds/src/main/java/io/github/arun0009/idempotent/rds/RdsConfig.java index 5718699..f5475a4 100644 --- a/idempotent-rds/src/main/java/io/github/arun0009/idempotent/rds/RdsConfig.java +++ b/idempotent-rds/src/main/java/io/github/arun0009/idempotent/rds/RdsConfig.java @@ -1,20 +1,30 @@ package io.github.arun0009.idempotent.rds; +import com.fasterxml.jackson.annotation.JsonTypeInfo; import io.github.arun0009.idempotent.core.persistence.IdempotentStore; -import org.springframework.beans.factory.annotation.Value; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; import org.springframework.boot.autoconfigure.AutoConfigureAfter; import org.springframework.boot.autoconfigure.condition.ConditionalOnBean; import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean; +import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty; +import org.springframework.boot.context.properties.EnableConfigurationProperties; import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration; import org.springframework.jdbc.core.JdbcTemplate; -import org.springframework.scheduling.annotation.EnableScheduling; +import org.springframework.scheduling.TaskScheduler; +import org.springframework.scheduling.concurrent.SimpleAsyncTaskScheduler; +import tools.jackson.databind.DefaultTyping; +import tools.jackson.databind.json.JsonMapper; +import tools.jackson.databind.jsontype.BasicPolymorphicTypeValidator; +import tools.jackson.databind.jsontype.impl.DefaultTypeResolverBuilder; import javax.sql.DataSource; +import java.time.Duration; @Configuration -@EnableScheduling @ConditionalOnBean(DataSource.class) +@EnableConfigurationProperties(RdsIdempotentProperties.class) @AutoConfigureAfter( name = { "org.springframework.boot.autoconfigure.jdbc.DataSourceAutoConfiguration", @@ -22,20 +32,54 @@ }) public class RdsConfig { + private static final Logger log = LoggerFactory.getLogger(RdsConfig.class); + + @Bean + @ConditionalOnMissingBean + public JsonMapper rdsJsonMapper() { + log.warn("Using an unrestricted polymorphic type validator for RDS idempotent store. " + + "Without restrictions of the PolymorphicTypeValidator, deserialization is " + + "vulnerable to arbitrary code execution when reading from untrusted sources."); + BasicPolymorphicTypeValidator ptv = BasicPolymorphicTypeValidator.builder() + .allowIfBaseType(Object.class) + .allowIfSubType((ctx, clazz) -> true) + .build(); + return JsonMapper.builder() + .polymorphicTypeValidator(ptv) + .setDefaultTyping(new DefaultTypeResolverBuilder( + ptv, DefaultTyping.NON_FINAL, JsonTypeInfo.As.PROPERTY, JsonTypeInfo.Id.CLASS, "@class")) + .build(); + } + @Bean @ConditionalOnMissingBean public IdempotentStore idempotentStore( - JdbcTemplate jdbcTemplate, @Value("${idempotent.rds.table.name:idempotent}") String tableName) { - return new RdsIdempotentStore(jdbcTemplate, tableName); + JdbcTemplate jdbcTemplate, RdsIdempotentProperties properties, JsonMapper rdsJsonMapper) { + return new RdsIdempotentStore(jdbcTemplate, properties.getTableName(), rdsJsonMapper); } @Bean @ConditionalOnMissingBean + @ConditionalOnProperty(prefix = "idempotent.rds.cleanup", name = "enabled", matchIfMissing = true) public RdsCleanupTask rdsCleanupTask( - JdbcTemplate jdbcTemplate, - @Value("${idempotent.rds.table.name:idempotent}") String tableName, - @Value("${idempotent.rds.cleanup.batch.size:1000}") int batchSize) { + JdbcTemplate jdbcTemplate, RdsIdempotentProperties properties, TaskScheduler rdsCleanupTaskScheduler) { RdsDialect dialect = RdsDialect.detect(jdbcTemplate); - return new RdsCleanupTask(jdbcTemplate, tableName, dialect, batchSize); + RdsCleanupTask cleanupTask = new RdsCleanupTask( + jdbcTemplate, + properties.getTableName(), + dialect, + properties.getCleanup().getBatchSize()); + rdsCleanupTaskScheduler.scheduleWithFixedDelay( + cleanupTask::cleanup, Duration.ofMillis(properties.getCleanup().getFixedDelay())); + return cleanupTask; + } + + @Bean + @ConditionalOnMissingBean(name = "rdsCleanupTaskScheduler") + @ConditionalOnProperty(prefix = "idempotent.rds.cleanup", name = "enabled", matchIfMissing = true) + public TaskScheduler rdsCleanupTaskScheduler() { + SimpleAsyncTaskScheduler scheduler = new SimpleAsyncTaskScheduler(); + scheduler.setThreadNamePrefix("idempotent-rds-cleanup-"); + return scheduler; } } diff --git a/idempotent-rds/src/main/java/io/github/arun0009/idempotent/rds/RdsDialect.java b/idempotent-rds/src/main/java/io/github/arun0009/idempotent/rds/RdsDialect.java index 884559c..a266ca5 100644 --- a/idempotent-rds/src/main/java/io/github/arun0009/idempotent/rds/RdsDialect.java +++ b/idempotent-rds/src/main/java/io/github/arun0009/idempotent/rds/RdsDialect.java @@ -1,5 +1,7 @@ package io.github.arun0009.idempotent.rds; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; import org.springframework.jdbc.core.JdbcTemplate; import java.sql.Connection; @@ -11,6 +13,8 @@ public enum RdsDialect { H2, GENERIC; + private static final Logger log = LoggerFactory.getLogger(RdsDialect.class); + public static RdsDialect detect(JdbcTemplate jdbcTemplate) { try { return jdbcTemplate.execute((Connection conn) -> { @@ -23,9 +27,14 @@ public static RdsDialect detect(JdbcTemplate jdbcTemplate) { } else if (databaseProductName.contains("h2")) { return H2; } + + log.warn( + "Unknown database product '{}', falling back to GENERIC dialect. SQL syntax may not be optimal.", + metaData.getDatabaseProductName()); return GENERIC; }); } catch (Exception e) { + log.warn("Failed to detect database dialect, falling back to GENERIC", e); return GENERIC; } } diff --git a/idempotent-rds/src/main/java/io/github/arun0009/idempotent/rds/RdsIdempotentProperties.java b/idempotent-rds/src/main/java/io/github/arun0009/idempotent/rds/RdsIdempotentProperties.java new file mode 100644 index 0000000..1e90f8c --- /dev/null +++ b/idempotent-rds/src/main/java/io/github/arun0009/idempotent/rds/RdsIdempotentProperties.java @@ -0,0 +1,62 @@ +package io.github.arun0009.idempotent.rds; + +import org.springframework.boot.context.properties.ConfigurationProperties; + +@ConfigurationProperties(prefix = "idempotent.rds") +public class RdsIdempotentProperties { + + /** Name of the database table for idempotent keys. */ + private String tableName = "idempotent"; + + private final Cleanup cleanup = new Cleanup(); + + public String getTableName() { + return tableName; + } + + public void setTableName(String tableName) { + this.tableName = tableName; + } + + public Cleanup getCleanup() { + return cleanup; + } + + public static class Cleanup { + + /** + * Whether the cleanup task is enabled. Set to false for CDS or AOT cache runs. + */ + private boolean enabled = true; + + /** Number of expired records to delete per batch. */ + private int batchSize = 1000; + + /** Fixed delay in milliseconds between cleanup runs. */ + private long fixedDelay = 60000; + + public boolean isEnabled() { + return enabled; + } + + public void setEnabled(boolean enabled) { + this.enabled = enabled; + } + + public int getBatchSize() { + return batchSize; + } + + public void setBatchSize(int batchSize) { + this.batchSize = batchSize; + } + + public long getFixedDelay() { + return fixedDelay; + } + + public void setFixedDelay(long fixedDelay) { + this.fixedDelay = fixedDelay; + } + } +} diff --git a/idempotent-rds/src/main/java/io/github/arun0009/idempotent/rds/RdsIdempotentStore.java b/idempotent-rds/src/main/java/io/github/arun0009/idempotent/rds/RdsIdempotentStore.java index b8cb9d3..94632a7 100644 --- a/idempotent-rds/src/main/java/io/github/arun0009/idempotent/rds/RdsIdempotentStore.java +++ b/idempotent-rds/src/main/java/io/github/arun0009/idempotent/rds/RdsIdempotentStore.java @@ -13,18 +13,20 @@ import java.sql.SQLException; /** - * RDS idempotent store using JdbcTemplate with atomic race condition protection. + * RDS idempotent store using JdbcTemplate with atomic race condition + * protection. */ public class RdsIdempotentStore implements IdempotentStore { private final JdbcTemplate jdbcTemplate; private final String tableName; - private final JsonMapper jsonMapper = JsonMapper.shared(); + private final JsonMapper jsonMapper; private final RdsDialect dialect; - public RdsIdempotentStore(JdbcTemplate jdbcTemplate, String tableName) { + public RdsIdempotentStore(JdbcTemplate jdbcTemplate, String tableName, JsonMapper jsonMapper) { this.jdbcTemplate = jdbcTemplate; this.tableName = tableName; + this.jsonMapper = jsonMapper; this.dialect = RdsDialect.detect(jdbcTemplate); } @@ -80,12 +82,10 @@ public void store(IdempotentKey key, Value value) { String responseJson = value.response() != null ? jsonMapper.writeValueAsString(value.response()) : null; long now = System.currentTimeMillis(); - if (dialect == RdsDialect.POSTGRES) { - storePostgres(key, value, responseJson, now); - } else if (dialect == RdsDialect.MYSQL) { - storeMySQL(key, value, responseJson, now); - } else { - storeGeneric(key, value, responseJson, now); + switch (dialect) { + case POSTGRES -> storePostgres(key, value, responseJson, now); + case MYSQL -> storeMySQL(key, value, responseJson, now); + case H2, GENERIC -> storeGeneric(key, value, responseJson, now); } } catch (JacksonException e) { diff --git a/idempotent-rds/src/test/java/io/github/arun0009/idempotent/rds/RdsIdempotentStoreH2Test.java b/idempotent-rds/src/test/java/io/github/arun0009/idempotent/rds/RdsIdempotentStoreH2Test.java index 1de33e7..dfccadb 100644 --- a/idempotent-rds/src/test/java/io/github/arun0009/idempotent/rds/RdsIdempotentStoreH2Test.java +++ b/idempotent-rds/src/test/java/io/github/arun0009/idempotent/rds/RdsIdempotentStoreH2Test.java @@ -14,6 +14,7 @@ import org.springframework.test.context.ContextConfiguration; import org.springframework.test.context.TestPropertySource; import org.springframework.test.context.junit.jupiter.SpringExtension; +import tools.jackson.databind.json.JsonMapper; import javax.sql.DataSource; @@ -28,7 +29,7 @@ */ @ExtendWith(SpringExtension.class) @ContextConfiguration(classes = RdsIdempotentStoreH2Test.H2TestConfig.class) -@TestPropertySource(properties = {"idempotent.rds.table.name=idempotent"}) +@TestPropertySource(properties = {"idempotent.rds.table-name=idempotent"}) public class RdsIdempotentStoreH2Test { @Autowired @@ -59,7 +60,7 @@ public JdbcTemplate jdbcTemplate(DataSource dataSource) { @Bean public IdempotentStore idempotentStore(JdbcTemplate jdbcTemplate) { - return new RdsIdempotentStore(jdbcTemplate, "idempotent"); + return new RdsIdempotentStore(jdbcTemplate, "idempotent", JsonMapper.shared()); } @Bean @@ -74,15 +75,15 @@ public RdsCleanupTask rdsCleanupTask(JdbcTemplate jdbcTemplate) { public void setUp() { // Create table for H2 jdbcTemplate.update(""" - CREATE TABLE IF NOT EXISTS idempotent ( - key_id VARCHAR(255) NOT NULL, - process_name VARCHAR(255) NOT NULL, - status VARCHAR(50), - expiration_time_millis BIGINT, - response TEXT, - PRIMARY KEY (key_id, process_name) - ) - """); + CREATE TABLE IF NOT EXISTS idempotent ( + key_id VARCHAR(255) NOT NULL, + process_name VARCHAR(255) NOT NULL, + status VARCHAR(50), + expiration_time_millis BIGINT, + response TEXT, + PRIMARY KEY (key_id, process_name) + ) + """); jdbcTemplate.update("CREATE INDEX IF NOT EXISTS idx_expiration_time ON idempotent(expiration_time_millis)"); jdbcTemplate.update("DELETE FROM idempotent"); diff --git a/idempotent-rds/src/test/java/io/github/arun0009/idempotent/rds/RdsIdempotentStoreMySQLTest.java b/idempotent-rds/src/test/java/io/github/arun0009/idempotent/rds/RdsIdempotentStoreMySQLTest.java index 5ca368d..eb6c189 100644 --- a/idempotent-rds/src/test/java/io/github/arun0009/idempotent/rds/RdsIdempotentStoreMySQLTest.java +++ b/idempotent-rds/src/test/java/io/github/arun0009/idempotent/rds/RdsIdempotentStoreMySQLTest.java @@ -21,7 +21,7 @@ @ExtendWith(SpringExtension.class) @ContextConfiguration(classes = RdsMySQLTestConfig.class) -@TestPropertySource(properties = {"idempotent.rds.table.name=idempotent"}) +@TestPropertySource(properties = {"idempotent.rds.table-name=idempotent"}) public class RdsIdempotentStoreMySQLTest { @Autowired @@ -37,15 +37,15 @@ public class RdsIdempotentStoreMySQLTest { public void setUp() { // Create the table for MySQL jdbcTemplate.update(""" - CREATE TABLE IF NOT EXISTS idempotent ( - key_id VARCHAR(255) NOT NULL, - process_name VARCHAR(255) NOT NULL, - status VARCHAR(50), - expiration_time_millis BIGINT, - response TEXT, - PRIMARY KEY (key_id, process_name) - ) - """); + CREATE TABLE IF NOT EXISTS idempotent ( + key_id VARCHAR(255) NOT NULL, + process_name VARCHAR(255) NOT NULL, + status VARCHAR(50), + expiration_time_millis BIGINT, + response TEXT, + PRIMARY KEY (key_id, process_name) + ) + """); // MySQL doesn't support CREATE INDEX IF NOT EXISTS, so we check first try { diff --git a/idempotent-rds/src/test/java/io/github/arun0009/idempotent/rds/RdsIdempotentStorePostgreSQLTest.java b/idempotent-rds/src/test/java/io/github/arun0009/idempotent/rds/RdsIdempotentStorePostgreSQLTest.java index fee6914..217e20d 100644 --- a/idempotent-rds/src/test/java/io/github/arun0009/idempotent/rds/RdsIdempotentStorePostgreSQLTest.java +++ b/idempotent-rds/src/test/java/io/github/arun0009/idempotent/rds/RdsIdempotentStorePostgreSQLTest.java @@ -21,7 +21,7 @@ @ExtendWith(SpringExtension.class) @ContextConfiguration(classes = RdsPostgreSQLTestConfig.class) -@TestPropertySource(properties = {"idempotent.rds.table.name=idempotent"}) +@TestPropertySource(properties = {"idempotent.rds.table-name=idempotent"}) public class RdsIdempotentStorePostgreSQLTest { @Autowired @@ -37,15 +37,15 @@ public class RdsIdempotentStorePostgreSQLTest { public void setUp() { // Create the table for PostgreSQL jdbcTemplate.update(""" - CREATE TABLE IF NOT EXISTS idempotent ( - key_id VARCHAR(255) NOT NULL, - process_name VARCHAR(255) NOT NULL, - status VARCHAR(50), - expiration_time_millis BIGINT, - response TEXT, - PRIMARY KEY (key_id, process_name) - ) - """); + CREATE TABLE IF NOT EXISTS idempotent ( + key_id VARCHAR(255) NOT NULL, + process_name VARCHAR(255) NOT NULL, + status VARCHAR(50), + expiration_time_millis BIGINT, + response TEXT, + PRIMARY KEY (key_id, process_name) + ) + """); jdbcTemplate.update("CREATE INDEX IF NOT EXISTS idx_expiration_time ON idempotent(expiration_time_millis)"); diff --git a/idempotent-rds/src/test/java/io/github/arun0009/idempotent/rds/RdsMySQLTestConfig.java b/idempotent-rds/src/test/java/io/github/arun0009/idempotent/rds/RdsMySQLTestConfig.java index 84c5542..7c1d527 100644 --- a/idempotent-rds/src/test/java/io/github/arun0009/idempotent/rds/RdsMySQLTestConfig.java +++ b/idempotent-rds/src/test/java/io/github/arun0009/idempotent/rds/RdsMySQLTestConfig.java @@ -11,6 +11,7 @@ import org.testcontainers.junit.jupiter.Container; import org.testcontainers.junit.jupiter.Testcontainers; import org.testcontainers.utility.DockerImageName; +import tools.jackson.databind.json.JsonMapper; import javax.sql.DataSource; @@ -46,7 +47,7 @@ public JdbcTemplate jdbcTemplate(DataSource dataSource) { @Bean public IdempotentStore idempotentStore(JdbcTemplate jdbcTemplate) { - return new RdsIdempotentStore(jdbcTemplate, "idempotent"); + return new RdsIdempotentStore(jdbcTemplate, "idempotent", JsonMapper.shared()); } @Bean diff --git a/idempotent-rds/src/test/java/io/github/arun0009/idempotent/rds/RdsPostgreSQLTestConfig.java b/idempotent-rds/src/test/java/io/github/arun0009/idempotent/rds/RdsPostgreSQLTestConfig.java index c602a40..22eec9e 100644 --- a/idempotent-rds/src/test/java/io/github/arun0009/idempotent/rds/RdsPostgreSQLTestConfig.java +++ b/idempotent-rds/src/test/java/io/github/arun0009/idempotent/rds/RdsPostgreSQLTestConfig.java @@ -11,6 +11,7 @@ import org.testcontainers.junit.jupiter.Container; import org.testcontainers.junit.jupiter.Testcontainers; import org.testcontainers.utility.DockerImageName; +import tools.jackson.databind.json.JsonMapper; import javax.sql.DataSource; @@ -46,7 +47,7 @@ public JdbcTemplate jdbcTemplate(DataSource dataSource) { @Bean public IdempotentStore idempotentStore(JdbcTemplate jdbcTemplate) { - return new RdsIdempotentStore(jdbcTemplate, "idempotent"); + return new RdsIdempotentStore(jdbcTemplate, "idempotent", JsonMapper.shared()); } @Bean From 5b4353b1fabcaafac51454281e4a5b7ed02f0995 Mon Sep 17 00:00:00 2001 From: Arun Gopalpuri Date: Sat, 21 Feb 2026 23:59:09 -0800 Subject: [PATCH 3/6] removing env as it has no properties --- .github/workflows/qodana_code_quality.yml | 2 -- 1 file changed, 2 deletions(-) diff --git a/.github/workflows/qodana_code_quality.yml b/.github/workflows/qodana_code_quality.yml index 16ba9ca..fc11cdb 100644 --- a/.github/workflows/qodana_code_quality.yml +++ b/.github/workflows/qodana_code_quality.yml @@ -29,8 +29,6 @@ jobs: - name: Qodana Scan uses: JetBrains/qodana-action@v2025.3 - env: - #QODANA_TOKEN: ${{ secrets.QODANA_TOKEN }} with: # When pr-mode is set to true, Qodana analyzes only the files that have been changed pr-mode: false From 50ecd8df70532657412fc0742fafd46e1e26af25 Mon Sep 17 00:00:00 2001 From: Arun Gopalpuri Date: Mon, 23 Feb 2026 00:52:36 -0800 Subject: [PATCH 4/6] simplify RDS implementation and revert store concurrency logic --- .../core/persistence/IdempotentStore.java | 6 +- .../persistence/InMemoryIdempotentStore.java | 20 +--- .../dynamo/DynamoIdempotentStore.java | 81 ++++------------ .../idempotent/nats/NatsIdempotentStore.java | 49 +++++----- .../nats/NoopNatsAutoConfiguration.java | 8 +- idempotent-rds/README.md | 11 ++- .../idempotent/rds/RdsIdempotentStore.java | 96 ++++--------------- .../rds/RdsIdempotentStoreH2Test.java | 10 +- .../rds/RdsIdempotentStoreMySQLTest.java | 11 ++- .../rds/RdsIdempotentStorePostgreSQLTest.java | 11 ++- .../redis/RedisIdempotentStore.java | 81 ++-------------- 11 files changed, 95 insertions(+), 289 deletions(-) diff --git a/idempotent-core/src/main/java/io/github/arun0009/idempotent/core/persistence/IdempotentStore.java b/idempotent-core/src/main/java/io/github/arun0009/idempotent/core/persistence/IdempotentStore.java index 0cc8750..565eb6f 100644 --- a/idempotent-core/src/main/java/io/github/arun0009/idempotent/core/persistence/IdempotentStore.java +++ b/idempotent-core/src/main/java/io/github/arun0009/idempotent/core/persistence/IdempotentStore.java @@ -54,11 +54,7 @@ record IdempotentKey(String key, String processName) implements Serializable {} * @param expirationTimeInMilliSeconds expiry time of idempotent entry from store * @param response this is response received from downstream apis. */ - record Value(String status, Long expirationTimeInMilliSeconds, Object response) implements Serializable { - public boolean isExpired() { - return expirationTimeInMilliSeconds != null && expirationTimeInMilliSeconds < System.currentTimeMillis(); - } - } + record Value(String status, Long expirationTimeInMilliSeconds, Object response) implements Serializable {} /** * The enum Status. diff --git a/idempotent-core/src/main/java/io/github/arun0009/idempotent/core/persistence/InMemoryIdempotentStore.java b/idempotent-core/src/main/java/io/github/arun0009/idempotent/core/persistence/InMemoryIdempotentStore.java index 48950da..dc10d82 100644 --- a/idempotent-core/src/main/java/io/github/arun0009/idempotent/core/persistence/InMemoryIdempotentStore.java +++ b/idempotent-core/src/main/java/io/github/arun0009/idempotent/core/persistence/InMemoryIdempotentStore.java @@ -23,14 +23,7 @@ public Value getValue(IdempotentKey idempotentKey, Class returnType) { @Override public void store(IdempotentKey idempotentKey, Value value) { - map.compute(idempotentKey, (k, v) -> { - if (v == null || v.isExpired()) { - return value; - } - // Race condition - key already exists and not expired - // Silent fail as per void interface contract - return v; - }); + map.put(idempotentKey, value); } @Override @@ -40,16 +33,7 @@ public void remove(IdempotentKey idempotentKey) { @Override public void update(IdempotentKey idempotentKey, Value value) { - map.compute(idempotentKey, (k, v) -> { - if (v == null - || v.isExpired() - || IdempotentStore.Status.INPROGRESS.name().equals(v.status())) { - return value; - } - // Race condition - key not in correct state for update - // Silent fail as per void interface contract - return v; - }); + map.replace(idempotentKey, value); } /** diff --git a/idempotent-dynamo/src/main/java/io/github/arun0009/idempotent/dynamo/DynamoIdempotentStore.java b/idempotent-dynamo/src/main/java/io/github/arun0009/idempotent/dynamo/DynamoIdempotentStore.java index bb2bf62..c7fb2e9 100644 --- a/idempotent-dynamo/src/main/java/io/github/arun0009/idempotent/dynamo/DynamoIdempotentStore.java +++ b/idempotent-dynamo/src/main/java/io/github/arun0009/idempotent/dynamo/DynamoIdempotentStore.java @@ -4,20 +4,14 @@ import io.github.arun0009.idempotent.core.persistence.IdempotentStore; import software.amazon.awssdk.enhanced.dynamodb.DynamoDbEnhancedClient; import software.amazon.awssdk.enhanced.dynamodb.DynamoDbTable; -import software.amazon.awssdk.enhanced.dynamodb.Expression; import software.amazon.awssdk.enhanced.dynamodb.Key; import software.amazon.awssdk.enhanced.dynamodb.TableSchema; import software.amazon.awssdk.enhanced.dynamodb.model.DeleteItemEnhancedRequest; -import software.amazon.awssdk.enhanced.dynamodb.model.PutItemEnhancedRequest; -import software.amazon.awssdk.services.dynamodb.model.AttributeValue; -import software.amazon.awssdk.services.dynamodb.model.ConditionalCheckFailedException; import tools.jackson.core.JacksonException; import tools.jackson.databind.json.JsonMapper; -import java.util.Map; - /** - * Dynamo idempotent store with conditional expressions for race condition handling + * Dynamo idempotent store. */ public class DynamoIdempotentStore implements IdempotentStore { @@ -54,14 +48,10 @@ public Value getValue(IdempotentKey idempotentKey, Class returnType) { if (idempotentItem == null) { return null; } - Value value = new Value( + return new Value( idempotentItem.getStatus(), idempotentItem.getExpirationTimeInMilliSeconds(), jsonMapper.readValue(idempotentItem.getResponse(), returnType)); - if (value.isExpired()) { - return null; - } - return value; } catch (JacksonException e) { throw new IdempotentException( String.format("Error getting response from dynamo for key: %s ", idempotentKey.key()), e); @@ -71,24 +61,14 @@ public Value getValue(IdempotentKey idempotentKey, Class returnType) { @Override public void store(IdempotentKey idempotentKey, Value value) { try { - IdempotentItem item = createItem(idempotentKey, value); - // Atomic condition: only insert if key doesn't exist or expired - Expression condition = Expression.builder() - .expression("attribute_not_exists(#key) OR expirationTimeInMilliSeconds < :now") - .expressionNames(Map.of("#key", "key")) - .expressionValues(Map.of( - ":now", - AttributeValue.builder() - .n(String.valueOf(System.currentTimeMillis())) - .build())) - .build(); - getTable() - .putItem(PutItemEnhancedRequest.builder(IdempotentItem.class) - .item(item) - .conditionExpression(condition) - .build()); - } catch (ConditionalCheckFailedException e) { - // Race condition - key exists and not expired, silent fail + DynamoDbTable idempotentTable = getTable(); + IdempotentItem idempotentItem = new IdempotentItem(); + idempotentItem.setKey(idempotentKey.key()); + idempotentItem.setProcessName(idempotentKey.processName()); + idempotentItem.setStatus(value.status()); + idempotentItem.setExpirationTimeInMilliSeconds(value.expirationTimeInMilliSeconds()); + idempotentItem.setResponse(jsonMapper.writeValueAsString(value.response())); + idempotentTable.putItem(idempotentItem); } catch (JacksonException e) { throw new IdempotentException("error storing idempotent item", e); } @@ -109,41 +89,16 @@ public void remove(IdempotentKey idempotentKey) { @Override public void update(IdempotentKey idempotentKey, Value value) { try { - IdempotentItem item = createItem(idempotentKey, value); - Expression condition = Expression.builder() - .expression( - "attribute_exists(#key) AND (#status = :inprogress OR expirationTimeInMilliSeconds < :now)") - .expressionNames(Map.of("#key", "key", "#status", "status")) - .expressionValues(Map.of( - ":now", - AttributeValue.builder() - .n(String.valueOf(System.currentTimeMillis())) - .build(), - ":inprogress", - AttributeValue.builder() - .s(IdempotentStore.Status.INPROGRESS.name()) - .build())) - .build(); - - getTable() - .putItem(PutItemEnhancedRequest.builder(IdempotentItem.class) - .item(item) - .conditionExpression(condition) - .build()); - } catch (ConditionalCheckFailedException e) { - // Race condition - key doesn't exist or not in correct state, silent fail + DynamoDbTable idempotentTable = getTable(); + IdempotentItem idempotentItem = new IdempotentItem(); + idempotentItem.setKey(idempotentKey.key()); + idempotentItem.setProcessName(idempotentKey.processName()); + idempotentItem.setStatus(value.status()); + idempotentItem.setExpirationTimeInMilliSeconds(value.expirationTimeInMilliSeconds()); + idempotentItem.setResponse(jsonMapper.writeValueAsString(value.response())); + idempotentTable.putItem(idempotentItem); } catch (JacksonException e) { throw new IdempotentException("error updating idempotent item", e); } } - - private IdempotentItem createItem(IdempotentKey idempotentKey, Value value) throws JacksonException { - IdempotentItem idempotentItem = new IdempotentItem(); - idempotentItem.setKey(idempotentKey.key()); - idempotentItem.setProcessName(idempotentKey.processName()); - idempotentItem.setStatus(value.status()); - idempotentItem.setExpirationTimeInMilliSeconds(value.expirationTimeInMilliSeconds()); - idempotentItem.setResponse(jsonMapper.writeValueAsString(value.response())); - return idempotentItem; - } } diff --git a/idempotent-nats/src/main/java/io/github/arun0009/idempotent/nats/NatsIdempotentStore.java b/idempotent-nats/src/main/java/io/github/arun0009/idempotent/nats/NatsIdempotentStore.java index 592864f..68cf7fb 100644 --- a/idempotent-nats/src/main/java/io/github/arun0009/idempotent/nats/NatsIdempotentStore.java +++ b/idempotent-nats/src/main/java/io/github/arun0009/idempotent/nats/NatsIdempotentStore.java @@ -11,6 +11,7 @@ import tools.jackson.databind.json.JsonMapper; import java.io.IOException; +import java.time.Duration; import java.time.Instant; import java.util.Base64; @@ -27,9 +28,8 @@ class NatsIdempotentStore implements IdempotentStore { } private static MessageTtl fromExpirationTimeInMs(long value) { - long ttlMillis = value - Instant.now().toEpochMilli(); - int ttlSeconds = (int) Math.max(1, (ttlMillis + 999) / 1000); // Round up, min 1 second - return MessageTtl.seconds(ttlSeconds); + int ttl = (int) Duration.ofMillis(value - Instant.now().toEpochMilli()).toSeconds(); + return MessageTtl.seconds(ttl + 1); } /** @@ -60,14 +60,7 @@ public Value getValue(IdempotentKey idemKey, Class returnType) { if (entry == null) return null; Wrappers.Value wrapperValue = mapper.readValue(entry.getValue(), Wrappers.Value.class); - Value value = wrapperValue.value(); - - // Check expiration using JVM time - if (value.isExpired()) { - return null; - } - - return value; + return wrapperValue.value(); } catch (IOException | JetStreamApiException e) { log.error("Error reading value from nats store", e); return null; @@ -83,16 +76,24 @@ public void store(IdempotentKey idemKey, Value value) { byte[] content = mapper.writeValueAsBytes(new Wrappers.Value(value)); MessageTtl messageTtl = fromExpirationTimeInMs(value.expirationTimeInMilliSeconds()); - // Try atomic create - will fail with 10071 if key already exists - kv.create(key, content, messageTtl); + + createOrUpdate(key, content, messageTtl); + } catch (IOException | JetStreamApiException e) { + throw new NatsIdempotentExceptions("Error storing value in nats", e); + } + } + + private void createOrUpdate(String key, byte[] value, MessageTtl ttl) throws JetStreamApiException, IOException { + try { + kv.create(key, value, ttl); } catch (JetStreamApiException e) { + // Wrong last sequence, the key already exists. if (e.getApiErrorCode() == 10071) { - // key already exists - silent fail (race condition) + kv.put(key, value); + log.atTrace().log("Updated key {}", key); return; } - throw new NatsIdempotentExceptions("Error storing value in nats", e); - } catch (IOException e) { - throw new NatsIdempotentExceptions("Error storing value in nats", e); + throw e; } } @@ -111,18 +112,12 @@ public void remove(IdempotentKey idemKey) { public void update(IdempotentKey idemKey, Value value) { try { log.atDebug().log("Updating key {} with status {}", idemKey, value.status()); + log.atTrace().log(value::toString); var key = encodeIfNotValid(idemKey); - KeyValueEntry entry = kv.get(key); - if (entry == null) return; // key doesn't exist, silent fail byte[] content = mapper.writeValueAsBytes(new Wrappers.Value(value)); - kv.update(key, content, entry.getRevision()); // Atomic CAS update - } catch (JetStreamApiException e) { - if (e.getApiErrorCode() == 10071) { - return; // CAS failed, silent fail - } - throw new NatsIdempotentExceptions("Error updating value in nats", e); - } catch (IOException e) { - throw new NatsIdempotentExceptions("Error updating value in nats", e); + kv.put(key, content); + } catch (IOException | JetStreamApiException e) { + throw new NatsIdempotentExceptions("Error storing value in nats", e); } } } diff --git a/idempotent-nats/src/main/java/io/github/arun0009/idempotent/nats/NoopNatsAutoConfiguration.java b/idempotent-nats/src/main/java/io/github/arun0009/idempotent/nats/NoopNatsAutoConfiguration.java index dadda2b..eef5c91 100644 --- a/idempotent-nats/src/main/java/io/github/arun0009/idempotent/nats/NoopNatsAutoConfiguration.java +++ b/idempotent-nats/src/main/java/io/github/arun0009/idempotent/nats/NoopNatsAutoConfiguration.java @@ -28,16 +28,12 @@ public Value getValue(IdempotentKey key, Class returnType) { } @Override - public void store(IdempotentKey key, Value value) { - // No-op store - } + public void store(IdempotentKey key, Value value) {} @Override public void remove(IdempotentKey key) {} @Override - public void update(IdempotentKey key, Value value) { - // No-op update - } + public void update(IdempotentKey key, Value value) {} } } diff --git a/idempotent-rds/README.md b/idempotent-rds/README.md index 3e921c5..0af36c8 100644 --- a/idempotent-rds/README.md +++ b/idempotent-rds/README.md @@ -59,13 +59,14 @@ See main [README](../README.md) for general idempotent configuration. Default Value: `1000` Description: Number of expired keys to delete in each batch to prevent long-running database locks. -## Distributed Systems Support +## Cleanup -This module is designed to work in a distributed environment with multiple application instances. +Unlike Redis or DynamoDb, RDS does not support native TTL for records. This module includes a `RdsCleanupTask` that: + +1. **Scheduled Execution**: Runs on a configurable schedule on all application instances. +2. **Batch Deletion**: Efficiently removes expired records in batches to avoid long-running locks. +3. **Safe Concurrency**: It is safe to run on multiple instances; database transactions ensure that a row is deleted exactly once. -1. **Atomic Operations**: It relies on the database's primary key constraints (`key_id`, `process_name`) to ensure that only one instance can successfully `INSERT` (lock) a specific key. -2. **Concurrency**: If multiple containers try to process the same request simultaneously, the database will reject duplicate inserts with a constraint violation. The application handles this by identifying it as a duplicate request. -3. **Cleanup**: The `RdsCleanupTask` runs on all instances by default. It executes batched `DELETE` statements for expired rows to prevent long-running database locks. This is safe to run concurrently as database transactions ensure consistency (deleting an already deleted row is a no-op). ## Dependencies diff --git a/idempotent-rds/src/main/java/io/github/arun0009/idempotent/rds/RdsIdempotentStore.java b/idempotent-rds/src/main/java/io/github/arun0009/idempotent/rds/RdsIdempotentStore.java index 94632a7..aa0fe75 100644 --- a/idempotent-rds/src/main/java/io/github/arun0009/idempotent/rds/RdsIdempotentStore.java +++ b/idempotent-rds/src/main/java/io/github/arun0009/idempotent/rds/RdsIdempotentStore.java @@ -2,7 +2,6 @@ import io.github.arun0009.idempotent.core.exception.IdempotentException; import io.github.arun0009.idempotent.core.persistence.IdempotentStore; -import org.springframework.dao.DuplicateKeyException; import org.springframework.dao.EmptyResultDataAccessException; import org.springframework.jdbc.core.JdbcTemplate; import org.springframework.jdbc.core.RowMapper; @@ -36,7 +35,7 @@ public Value getValue(IdempotentKey key, Class returnType) { SELECT status, expiration_time_millis, response FROM %s WHERE key_id = ? AND process_name = ? """.formatted(tableName); try { - Value value = jdbcTemplate.queryForObject( + return jdbcTemplate.queryForObject( sql, new RowMapper() { @Override @@ -52,12 +51,7 @@ public Value mapRow(ResultSet rs, int rowNum) throws SQLException { if (responseJson != null) { response = jsonMapper.readValue(responseJson, returnType); } - Value result = new Value(status, expirationTime, response); - // Check expiration using JVM time - if (result.isExpired()) { - return null; // Treat expired as not found - } - return result; + return new Value(status, expirationTime, response); } catch (JacksonException e) { throw new IdempotentException("Error deserializing response", e); } @@ -65,12 +59,6 @@ public Value mapRow(ResultSet rs, int rowNum) throws SQLException { }, key.key(), key.processName()); - - // Additional expiration check - if (value != null && value.isExpired()) { - return null; - } - return value; } catch (EmptyResultDataAccessException e) { return null; } @@ -80,12 +68,11 @@ public Value mapRow(ResultSet rs, int rowNum) throws SQLException { public void store(IdempotentKey key, Value value) { try { String responseJson = value.response() != null ? jsonMapper.writeValueAsString(value.response()) : null; - long now = System.currentTimeMillis(); switch (dialect) { - case POSTGRES -> storePostgres(key, value, responseJson, now); - case MYSQL -> storeMySQL(key, value, responseJson, now); - case H2, GENERIC -> storeGeneric(key, value, responseJson, now); + case POSTGRES -> storePostgres(key, value, responseJson); + case MYSQL -> storeMySQL(key, value, responseJson); + case H2, GENERIC -> storeGeneric(key, value, responseJson); } } catch (JacksonException e) { @@ -93,78 +80,29 @@ public void store(IdempotentKey key, Value value) { } } - private void storePostgres(IdempotentKey key, Value value, String responseJson, long now) { - // Postgres: use ON CONFLICT with conditional WHERE clause + private void storePostgres(IdempotentKey key, Value value, String responseJson) { String sql = """ INSERT INTO %s (key_id, process_name, status, expiration_time_millis, response) VALUES (?, ?, ?, ?, ?) - ON CONFLICT (key_id, process_name) - DO UPDATE SET - status = EXCLUDED.status, - expiration_time_millis = EXCLUDED.expiration_time_millis, - response = EXCLUDED.response - WHERE %s.expiration_time_millis < ? - """.formatted(tableName, tableName); + """.formatted(tableName); jdbcTemplate.update( - sql, - key.key(), - key.processName(), - value.status(), - value.expirationTimeInMilliSeconds(), - responseJson, - now); + sql, key.key(), key.processName(), value.status(), value.expirationTimeInMilliSeconds(), responseJson); } - private void storeMySQL(IdempotentKey key, Value value, String responseJson, long now) { - // MySQL: use ON DUPLICATE KEY UPDATE with IF() condition - String insertSql = """ + private void storeMySQL(IdempotentKey key, Value value, String responseJson) { + String sql = """ INSERT INTO %s (key_id, process_name, status, expiration_time_millis, response) VALUES (?, ?, ?, ?, ?) - ON DUPLICATE KEY UPDATE - status = IF(expiration_time_millis < ?, VALUES(status), status), - expiration_time_millis = IF(expiration_time_millis < ?, VALUES(expiration_time_millis), expiration_time_millis), - response = IF(expiration_time_millis < ?, VALUES(response), response) """.formatted(tableName); jdbcTemplate.update( - insertSql, - key.key(), - key.processName(), - value.status(), - value.expirationTimeInMilliSeconds(), - responseJson, - now, - now, - now); + sql, key.key(), key.processName(), value.status(), value.expirationTimeInMilliSeconds(), responseJson); } - private void storeGeneric(IdempotentKey key, Value value, String response, long now) { - // Generic H2: INSERT first, if duplicate then try UPDATE with WHERE clause - try { - String insertSQL = """ - INSERT INTO %s (key_id, process_name, status, expiration_time_millis, response) VALUES (?, ?, ?, ?, ?) - """.formatted(tableName); - jdbcTemplate.update( - insertSQL, - key.key(), - key.processName(), - value.status(), - value.expirationTimeInMilliSeconds(), - response); - } catch (DuplicateKeyException e) { - // Key exists, try to update only if expired - String updateSql = """ - UPDATE %s SET status = ?, expiration_time_millis = ?, response = ? - WHERE key_id = ? AND process_name = ? AND expiration_time_millis < ? - """.formatted(tableName); - jdbcTemplate.update( - updateSql, - value.status(), - value.expirationTimeInMilliSeconds(), - response, - key.key(), - key.processName(), - now); - // IF UPDATE affects 0 rows, key exists but not expired - silent fail - } + private void storeGeneric(IdempotentKey key, Value value, String response) { + String sql = """ + INSERT INTO %s (key_id, process_name, status, expiration_time_millis, response) VALUES (?, ?, ?, ?, ?) + """.formatted(tableName); + jdbcTemplate.update( + sql, key.key(), key.processName(), value.status(), value.expirationTimeInMilliSeconds(), response); } @Override diff --git a/idempotent-rds/src/test/java/io/github/arun0009/idempotent/rds/RdsIdempotentStoreH2Test.java b/idempotent-rds/src/test/java/io/github/arun0009/idempotent/rds/RdsIdempotentStoreH2Test.java index dfccadb..41f67f2 100644 --- a/idempotent-rds/src/test/java/io/github/arun0009/idempotent/rds/RdsIdempotentStoreH2Test.java +++ b/idempotent-rds/src/test/java/io/github/arun0009/idempotent/rds/RdsIdempotentStoreH2Test.java @@ -10,6 +10,7 @@ import org.springframework.boot.jdbc.DataSourceBuilder; import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration; +import org.springframework.dao.DuplicateKeyException; import org.springframework.jdbc.core.JdbcTemplate; import org.springframework.test.context.ContextConfiguration; import org.springframework.test.context.TestPropertySource; @@ -20,7 +21,7 @@ import static org.junit.jupiter.api.Assertions.assertEquals; import static org.junit.jupiter.api.Assertions.assertNotNull; -import static org.junit.jupiter.api.Assertions.assertNull; +import static org.junit.jupiter.api.Assertions.assertThrows; /** * Fast H2-based tests for development. @@ -110,18 +111,17 @@ public void testDialectDetection() { } @Test - public void testRaceConditionProtection() { + public void testDuplicateStoreThrowsException() { IdempotentKey key = new IdempotentKey("race-key", "race-process"); Value value1 = new Value("INPROGRESS", System.currentTimeMillis() + 10000, null); idempotentStore.store(key, value1); Value value2 = new Value("COMPLETED", System.currentTimeMillis() + 20000, "overwritten"); - // This should NOT overwrite value1 due to race condition protection - idempotentStore.store(key, value2); + // This should throw DuplicateKeyException + assertThrows(DuplicateKeyException.class, () -> idempotentStore.store(key, value2)); Value retrieved = idempotentStore.getValue(key, String.class); assertNotNull(retrieved); assertEquals("INPROGRESS", retrieved.status()); // Should still be original - assertNull(retrieved.response()); // Original had null response } } diff --git a/idempotent-rds/src/test/java/io/github/arun0009/idempotent/rds/RdsIdempotentStoreMySQLTest.java b/idempotent-rds/src/test/java/io/github/arun0009/idempotent/rds/RdsIdempotentStoreMySQLTest.java index eb6c189..c60a7da 100644 --- a/idempotent-rds/src/test/java/io/github/arun0009/idempotent/rds/RdsIdempotentStoreMySQLTest.java +++ b/idempotent-rds/src/test/java/io/github/arun0009/idempotent/rds/RdsIdempotentStoreMySQLTest.java @@ -7,6 +7,7 @@ import org.junit.jupiter.api.Test; import org.junit.jupiter.api.extension.ExtendWith; import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.dao.DuplicateKeyException; import org.springframework.jdbc.core.JdbcTemplate; import org.springframework.test.context.ContextConfiguration; import org.springframework.test.context.TestPropertySource; @@ -18,6 +19,7 @@ import static org.junit.jupiter.api.Assertions.assertEquals; import static org.junit.jupiter.api.Assertions.assertNotNull; import static org.junit.jupiter.api.Assertions.assertNull; +import static org.junit.jupiter.api.Assertions.assertThrows; @ExtendWith(SpringExtension.class) @ContextConfiguration(classes = RdsMySQLTestConfig.class) @@ -90,20 +92,19 @@ public void testUpdate() { } @Test - public void testDuplicateStoreDoesNotOverwrite() { + public void testDuplicateStoreThrowsException() { IdempotentKey key = new IdempotentKey("dup-key", "dup-process"); Value value1 = new Value("INPROGRESS", System.currentTimeMillis() + 10000, null); idempotentStore.store(key, value1); Value value2 = new Value("COMPLETED", System.currentTimeMillis() + 20000, Map.of("data", "overwritten")); - // With race condition protection, this should NOT overwrite value1 - idempotentStore.store(key, value2); + // With strict insert, this should throw DuplicateKeyException + assertThrows(DuplicateKeyException.class, () -> idempotentStore.store(key, value2)); Value retrieved = idempotentStore.getValue(key, Map.class); assertNotNull(retrieved); - assertEquals("INPROGRESS", retrieved.status()); // Should still be the original value - assertNull(retrieved.response()); // Original value had null response + assertEquals("INPROGRESS", retrieved.status()); // Should still be original } @Test diff --git a/idempotent-rds/src/test/java/io/github/arun0009/idempotent/rds/RdsIdempotentStorePostgreSQLTest.java b/idempotent-rds/src/test/java/io/github/arun0009/idempotent/rds/RdsIdempotentStorePostgreSQLTest.java index 217e20d..16996e9 100644 --- a/idempotent-rds/src/test/java/io/github/arun0009/idempotent/rds/RdsIdempotentStorePostgreSQLTest.java +++ b/idempotent-rds/src/test/java/io/github/arun0009/idempotent/rds/RdsIdempotentStorePostgreSQLTest.java @@ -7,6 +7,7 @@ import org.junit.jupiter.api.Test; import org.junit.jupiter.api.extension.ExtendWith; import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.dao.DuplicateKeyException; import org.springframework.jdbc.core.JdbcTemplate; import org.springframework.test.context.ContextConfiguration; import org.springframework.test.context.TestPropertySource; @@ -18,6 +19,7 @@ import static org.junit.jupiter.api.Assertions.assertEquals; import static org.junit.jupiter.api.Assertions.assertNotNull; import static org.junit.jupiter.api.Assertions.assertNull; +import static org.junit.jupiter.api.Assertions.assertThrows; @ExtendWith(SpringExtension.class) @ContextConfiguration(classes = RdsPostgreSQLTestConfig.class) @@ -83,20 +85,19 @@ public void testUpdate() { } @Test - public void testDuplicateStoreDoesNotOverwrite() { + public void testDuplicateStoreThrowsException() { IdempotentKey key = new IdempotentKey("dup-key", "dup-process"); Value value1 = new Value("INPROGRESS", System.currentTimeMillis() + 10000, null); idempotentStore.store(key, value1); Value value2 = new Value("COMPLETED", System.currentTimeMillis() + 20000, Map.of("data", "overwritten")); - // With race condition protection, this should NOT overwrite value1 - idempotentStore.store(key, value2); + // With strict insert, this should throw DuplicateKeyException + assertThrows(DuplicateKeyException.class, () -> idempotentStore.store(key, value2)); Value retrieved = idempotentStore.getValue(key, Map.class); assertNotNull(retrieved); - assertEquals("INPROGRESS", retrieved.status()); // Should still be the original value - assertNull(retrieved.response()); // Original value had null response + assertEquals("INPROGRESS", retrieved.status()); // Should still be original } @Test diff --git a/idempotent-redis/src/main/java/io/github/arun0009/idempotent/redis/RedisIdempotentStore.java b/idempotent-redis/src/main/java/io/github/arun0009/idempotent/redis/RedisIdempotentStore.java index 6219e95..361fa38 100644 --- a/idempotent-redis/src/main/java/io/github/arun0009/idempotent/redis/RedisIdempotentStore.java +++ b/idempotent-redis/src/main/java/io/github/arun0009/idempotent/redis/RedisIdempotentStore.java @@ -3,69 +3,15 @@ import io.github.arun0009.idempotent.core.persistence.IdempotentStore; import org.springframework.data.redis.core.RedisTemplate; -import java.util.Collections; +import java.util.concurrent.TimeUnit; /** - * Redis idempotent store with atomic operations using Lua scripts + * Redis idempotent store */ public class RedisIdempotentStore implements IdempotentStore { private final RedisTemplate redisTemplate; - private static final String STORE_SCRIPT = """ - local existing = redis.call('GET', KEYS[1]) - local now = tonumber(ARGV[3]) - if not now then - now = 0 - end - - if not existing then - -- Key doesn't exist, safe to store it - local ttl = tonumber(ARGV[2]) - if ttl and type(ttl) == "number" and ttl > 0 and ttl ~= -1 then - redis.call('SET', KEYS[1], ARGV[1], 'PX', ttl) - else - redis.call('SET', KEYS[1], ARGV[1]) - end - return '"1"' - end - - -- Key exists check if expired - local success, val = pcall(cjson.decode, existing) - if success and val and val.expirationTimeInMilliSeconds and val.expirationTimeInMilliSeconds < now then - -- Expired, safe to overwrite - local ttl = tonumber(ARGV[2]) - if ttl and type(ttl) == "number" and ttl > 0 and ttl ~= -1 then - redis.call('SET', KEYS[1], ARGV[1], 'PX', ttl) - else - redis.call('SET', KEYS[1], ARGV[1]) - end - return '"1"' - end - -- Key exists and not expired, dont overwrite - return '"0"'"""; - - private static final String UPDATE_SCRIPT = """ - local existing = redis.call('GET', KEYS[1]) - if not existing then - -- Key doesn't exist, nothing to update - return '"0"' - end - - local success, val = pcall(cjson.decode, existing) - if success and val and val.status == 'INPROGRESS' then - -- Only update if status is INPROGRESS - local ttl = tonumber(ARGV[2]) - if ttl and type(ttl) == "number" and ttl > 0 and ttl ~= -1 then - redis.call('SET', KEYS[1], ARGV[1], 'PX', ttl) - else - redis.call('SET', KEYS[1], ARGV[1]) - end - return '"1"' - end - -- Key exists and not in progress, dont update - return '"0"'"""; - /** * Instantiates a new Redis idempotent store. * @@ -75,19 +21,6 @@ public RedisIdempotentStore(RedisTemplate redisTemplate) { this.redisTemplate = redisTemplate; } - private void executeRedisScript(String script, IdempotentKey key, Value value) { - long now = System.currentTimeMillis(); - long ttl = value.expirationTimeInMilliSeconds() != null ? value.expirationTimeInMilliSeconds() - now : -1; - - redisTemplate.execute( - new org.springframework.data.redis.core.script.DefaultRedisScript<>(script, String.class), - Collections.singletonList(key), - value, - String.valueOf(ttl), - String.valueOf(now)); - // Silent fail - race condition handled by Lua script - } - @Override public Value getValue(IdempotentKey key, Class returnType) { return redisTemplate.opsForValue().get(key); @@ -95,7 +28,10 @@ public Value getValue(IdempotentKey key, Class returnType) { @Override public void store(IdempotentKey key, Value value) { - executeRedisScript(STORE_SCRIPT, key, value); + redisTemplate.opsForValue().set(key, value); + if (value.expirationTimeInMilliSeconds() != null) { + redisTemplate.expire(key, value.expirationTimeInMilliSeconds(), TimeUnit.MILLISECONDS); + } } @Override @@ -105,6 +41,9 @@ public void remove(IdempotentKey key) { @Override public void update(IdempotentKey key, Value value) { - executeRedisScript(UPDATE_SCRIPT, key, value); + redisTemplate.opsForValue().set(key, value); + if (value.expirationTimeInMilliSeconds() != null) { + redisTemplate.expire(key, value.expirationTimeInMilliSeconds(), TimeUnit.MILLISECONDS); + } } } From aa280f143b6b427e8419e3367160385722eaff64 Mon Sep 17 00:00:00 2001 From: Arun Gopalpuri Date: Mon, 23 Feb 2026 21:35:51 -0800 Subject: [PATCH 5/6] optimizing rds module for lib + addressing pr comments --- idempotent-rds/README.md | 16 +++-- idempotent-rds/pom.xml | 17 ++++- ...sConfig.java => RdsAutoConfiguration.java} | 69 +++++++++++-------- .../idempotent/rds/RdsCleanupTask.java | 4 +- .../rds/RdsIdempotentProperties.java | 64 +++-------------- .../idempotent/rds/RdsIdempotentStore.java | 16 ++--- .../rds/RdsJacksonJsonBuilderCustomizer.java | 17 +++++ ...ot.autoconfigure.AutoConfiguration.imports | 2 +- .../rds/RdsIdempotentStoreH2Test.java | 14 ++-- .../rds/RdsIdempotentStoreMySQLTest.java | 4 +- .../rds/RdsIdempotentStorePostgreSQLTest.java | 4 +- .../idempotent/rds/RdsMySQLTestConfig.java | 14 ++-- .../rds/RdsPostgreSQLTestConfig.java | 14 ++-- pom.xml | 25 ++----- 14 files changed, 139 insertions(+), 141 deletions(-) rename idempotent-rds/src/main/java/io/github/arun0009/idempotent/rds/{RdsConfig.java => RdsAutoConfiguration.java} (55%) create mode 100644 idempotent-rds/src/main/java/io/github/arun0009/idempotent/rds/RdsJacksonJsonBuilderCustomizer.java diff --git a/idempotent-rds/README.md b/idempotent-rds/README.md index 0af36c8..2becf22 100644 --- a/idempotent-rds/README.md +++ b/idempotent-rds/README.md @@ -43,19 +43,19 @@ See main [README](../README.md) for general idempotent configuration. * Table Name - Property: `idempotent.rds.table.name` + Property: `idempotent.rds.table-name` Default Value: `idempotent` Description: The name of the database table to use. * Cleanup Schedule - Property: `idempotent.rds.cleanup.fixedDelay` + Property: `idempotent.rds.cleanup.fixed-delay` Default Value: `60000` (1 minute) Description: Fixed delay in milliseconds for the cleanup task that removes expired keys. * Cleanup Batch Size - Property: `idempotent.rds.cleanup.batch.size` + Property: `idempotent.rds.cleanup.batch-size` Default Value: `1000` Description: Number of expired keys to delete in each batch to prevent long-running database locks. @@ -70,7 +70,15 @@ Unlike Redis or DynamoDb, RDS does not support native TTL for records. This modu ## Dependencies -This module relies on `spring-boot-starter-jdbc`. You must configure your `DataSource` as per standard Spring Boot configuration (e.g., `spring.datasource.url`, etc.). +This module provides the core JDBC logic but does **not** bundle a connection pool or database drivers. You must provide them in your application: + +**The Recommended Way (via Spring Boot Starter):** +1. Add `spring-boot-starter-jdbc` to your application (this automatically provides `HikariCP` and configures the `DataSource` bean). +2. Add the JDBC driver for your database (e.g., `mysql-connector-j` or `postgresql`). +3. Configure your `DataSource` as per standard Spring Boot configuration (e.g., `spring.datasource.url`). + +**Manual Setup:** +If you prefer not to use the starter, you must manually provide a `DataSource` bean, a connection pool library (like `HikariCP`), and your database driver. ## Performance Tuning diff --git a/idempotent-rds/pom.xml b/idempotent-rds/pom.xml index f12d00c..aa87125 100644 --- a/idempotent-rds/pom.xml +++ b/idempotent-rds/pom.xml @@ -18,9 +18,18 @@ io.github.arun0009 idempotent-core + + org.springframework + spring-jdbc + + + org.springframework + spring-tx + org.springframework.boot - spring-boot-starter-jdbc + spring-boot-autoconfigure + provided org.springframework.boot @@ -34,11 +43,15 @@ spring-boot-starter-test test + + com.zaxxer + HikariCP + test + org.testcontainers junit-jupiter - ${testcontainers.version} test diff --git a/idempotent-rds/src/main/java/io/github/arun0009/idempotent/rds/RdsConfig.java b/idempotent-rds/src/main/java/io/github/arun0009/idempotent/rds/RdsAutoConfiguration.java similarity index 55% rename from idempotent-rds/src/main/java/io/github/arun0009/idempotent/rds/RdsConfig.java rename to idempotent-rds/src/main/java/io/github/arun0009/idempotent/rds/RdsAutoConfiguration.java index f5475a4..b7b6b22 100644 --- a/idempotent-rds/src/main/java/io/github/arun0009/idempotent/rds/RdsConfig.java +++ b/idempotent-rds/src/main/java/io/github/arun0009/idempotent/rds/RdsAutoConfiguration.java @@ -4,16 +4,17 @@ import io.github.arun0009.idempotent.core.persistence.IdempotentStore; import org.slf4j.Logger; import org.slf4j.LoggerFactory; -import org.springframework.boot.autoconfigure.AutoConfigureAfter; +import org.springframework.boot.autoconfigure.AutoConfiguration; import org.springframework.boot.autoconfigure.condition.ConditionalOnBean; +import org.springframework.boot.autoconfigure.condition.ConditionalOnClass; import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean; import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty; import org.springframework.boot.context.properties.EnableConfigurationProperties; import org.springframework.context.annotation.Bean; -import org.springframework.context.annotation.Configuration; import org.springframework.jdbc.core.JdbcTemplate; import org.springframework.scheduling.TaskScheduler; import org.springframework.scheduling.concurrent.SimpleAsyncTaskScheduler; +import org.springframework.util.Assert; import tools.jackson.databind.DefaultTyping; import tools.jackson.databind.json.JsonMapper; import tools.jackson.databind.jsontype.BasicPolymorphicTypeValidator; @@ -22,40 +23,45 @@ import javax.sql.DataSource; import java.time.Duration; -@Configuration -@ConditionalOnBean(DataSource.class) -@EnableConfigurationProperties(RdsIdempotentProperties.class) -@AutoConfigureAfter( - name = { +@AutoConfiguration( + afterName = { "org.springframework.boot.autoconfigure.jdbc.DataSourceAutoConfiguration", "org.springframework.boot.autoconfigure.jdbc.JdbcTemplateAutoConfiguration" }) -public class RdsConfig { +@ConditionalOnClass({DataSource.class, JdbcTemplate.class}) +@ConditionalOnBean(DataSource.class) +@EnableConfigurationProperties(RdsIdempotentProperties.class) +public class RdsAutoConfiguration { - private static final Logger log = LoggerFactory.getLogger(RdsConfig.class); + private static final Logger log = LoggerFactory.getLogger(RdsAutoConfiguration.class); @Bean @ConditionalOnMissingBean - public JsonMapper rdsJsonMapper() { - log.warn("Using an unrestricted polymorphic type validator for RDS idempotent store. " - + "Without restrictions of the PolymorphicTypeValidator, deserialization is " - + "vulnerable to arbitrary code execution when reading from untrusted sources."); - BasicPolymorphicTypeValidator ptv = BasicPolymorphicTypeValidator.builder() - .allowIfBaseType(Object.class) - .allowIfSubType((ctx, clazz) -> true) - .build(); - return JsonMapper.builder() - .polymorphicTypeValidator(ptv) - .setDefaultTyping(new DefaultTypeResolverBuilder( - ptv, DefaultTyping.NON_FINAL, JsonTypeInfo.As.PROPERTY, JsonTypeInfo.Id.CLASS, "@class")) - .build(); + public IdempotentStore idempotentStore( + JdbcTemplate jdbcTemplate, RdsIdempotentProperties properties, RdsJacksonJsonBuilderCustomizer customizer) { + Assert.hasText(properties.tableName(), "idempotent.rds.table-name must not be blank"); + + var jsonBuilder = JsonMapper.builder(); + customizer.customize(jsonBuilder); + + return new RdsIdempotentStore(jdbcTemplate, properties.tableName(), jsonBuilder.build()); } @Bean @ConditionalOnMissingBean - public IdempotentStore idempotentStore( - JdbcTemplate jdbcTemplate, RdsIdempotentProperties properties, JsonMapper rdsJsonMapper) { - return new RdsIdempotentStore(jdbcTemplate, properties.getTableName(), rdsJsonMapper); + public RdsJacksonJsonBuilderCustomizer rdsJacksonJsonBuilderCustomizer() { + return builder -> { + log.warn("Using an unrestricted polymorphic type validator for RDS idempotent store. " + + "Without restrictions of the PolymorphicTypeValidator, deserialization is " + + "vulnerable to arbitrary code execution when reading from untrusted sources."); + var ptv = BasicPolymorphicTypeValidator.builder() + .allowIfBaseType(Object.class) + .allowIfSubType((ctx, clazz) -> true) + .build(); + builder.polymorphicTypeValidator(ptv) + .setDefaultTyping(new DefaultTypeResolverBuilder( + ptv, DefaultTyping.NON_FINAL, JsonTypeInfo.As.PROPERTY, JsonTypeInfo.Id.CLASS, "@class")); + }; } @Bean @@ -63,14 +69,17 @@ public IdempotentStore idempotentStore( @ConditionalOnProperty(prefix = "idempotent.rds.cleanup", name = "enabled", matchIfMissing = true) public RdsCleanupTask rdsCleanupTask( JdbcTemplate jdbcTemplate, RdsIdempotentProperties properties, TaskScheduler rdsCleanupTaskScheduler) { + Assert.isTrue(properties.cleanup().batchSize() > 0, "idempotent.rds.cleanup.batch-size must be positive"); + Assert.isTrue(properties.cleanup().fixedDelay() > 0, "idempotent.rds.cleanup.fixed-delay must be positive"); + RdsDialect dialect = RdsDialect.detect(jdbcTemplate); - RdsCleanupTask cleanupTask = new RdsCleanupTask( + var cleanupTask = new RdsCleanupTask( jdbcTemplate, - properties.getTableName(), + properties.tableName(), dialect, - properties.getCleanup().getBatchSize()); + properties.cleanup().batchSize()); rdsCleanupTaskScheduler.scheduleWithFixedDelay( - cleanupTask::cleanup, Duration.ofMillis(properties.getCleanup().getFixedDelay())); + cleanupTask::cleanup, Duration.ofMillis(properties.cleanup().fixedDelay())); return cleanupTask; } @@ -78,7 +87,7 @@ public RdsCleanupTask rdsCleanupTask( @ConditionalOnMissingBean(name = "rdsCleanupTaskScheduler") @ConditionalOnProperty(prefix = "idempotent.rds.cleanup", name = "enabled", matchIfMissing = true) public TaskScheduler rdsCleanupTaskScheduler() { - SimpleAsyncTaskScheduler scheduler = new SimpleAsyncTaskScheduler(); + var scheduler = new SimpleAsyncTaskScheduler(); scheduler.setThreadNamePrefix("idempotent-rds-cleanup-"); return scheduler; } diff --git a/idempotent-rds/src/main/java/io/github/arun0009/idempotent/rds/RdsCleanupTask.java b/idempotent-rds/src/main/java/io/github/arun0009/idempotent/rds/RdsCleanupTask.java index 2600aee..a73266a 100644 --- a/idempotent-rds/src/main/java/io/github/arun0009/idempotent/rds/RdsCleanupTask.java +++ b/idempotent-rds/src/main/java/io/github/arun0009/idempotent/rds/RdsCleanupTask.java @@ -20,8 +20,8 @@ public RdsCleanupTask(JdbcTemplate jdbcTemplate, String tableName, RdsDialect di } public void cleanup() { - long now = System.currentTimeMillis(); - int totalDeleted = 0; + var now = System.currentTimeMillis(); + var totalDeleted = 0; int batchDeleted; do { diff --git a/idempotent-rds/src/main/java/io/github/arun0009/idempotent/rds/RdsIdempotentProperties.java b/idempotent-rds/src/main/java/io/github/arun0009/idempotent/rds/RdsIdempotentProperties.java index 1e90f8c..cba70fe 100644 --- a/idempotent-rds/src/main/java/io/github/arun0009/idempotent/rds/RdsIdempotentProperties.java +++ b/idempotent-rds/src/main/java/io/github/arun0009/idempotent/rds/RdsIdempotentProperties.java @@ -1,62 +1,20 @@ package io.github.arun0009.idempotent.rds; import org.springframework.boot.context.properties.ConfigurationProperties; +import org.springframework.boot.context.properties.bind.DefaultValue; @ConfigurationProperties(prefix = "idempotent.rds") -public class RdsIdempotentProperties { +public record RdsIdempotentProperties( + /** Name of the database table for idempotent keys. */ + @DefaultValue("idempotent") String tableName, Cleanup cleanup) { - /** Name of the database table for idempotent keys. */ - private String tableName = "idempotent"; + public record Cleanup( + /** Whether the cleanup task is enabled. Set to false for CDS or AOT cache runs. */ + @DefaultValue("true") boolean enabled, - private final Cleanup cleanup = new Cleanup(); + /** Number of expired records to delete per batch. */ + @DefaultValue("1000") int batchSize, - public String getTableName() { - return tableName; - } - - public void setTableName(String tableName) { - this.tableName = tableName; - } - - public Cleanup getCleanup() { - return cleanup; - } - - public static class Cleanup { - - /** - * Whether the cleanup task is enabled. Set to false for CDS or AOT cache runs. - */ - private boolean enabled = true; - - /** Number of expired records to delete per batch. */ - private int batchSize = 1000; - - /** Fixed delay in milliseconds between cleanup runs. */ - private long fixedDelay = 60000; - - public boolean isEnabled() { - return enabled; - } - - public void setEnabled(boolean enabled) { - this.enabled = enabled; - } - - public int getBatchSize() { - return batchSize; - } - - public void setBatchSize(int batchSize) { - this.batchSize = batchSize; - } - - public long getFixedDelay() { - return fixedDelay; - } - - public void setFixedDelay(long fixedDelay) { - this.fixedDelay = fixedDelay; - } - } + /** Fixed delay in milliseconds between cleanup runs. */ + @DefaultValue("60000") long fixedDelay) {} } diff --git a/idempotent-rds/src/main/java/io/github/arun0009/idempotent/rds/RdsIdempotentStore.java b/idempotent-rds/src/main/java/io/github/arun0009/idempotent/rds/RdsIdempotentStore.java index aa0fe75..0e5aa68 100644 --- a/idempotent-rds/src/main/java/io/github/arun0009/idempotent/rds/RdsIdempotentStore.java +++ b/idempotent-rds/src/main/java/io/github/arun0009/idempotent/rds/RdsIdempotentStore.java @@ -31,7 +31,7 @@ public RdsIdempotentStore(JdbcTemplate jdbcTemplate, String tableName, JsonMappe @Override public Value getValue(IdempotentKey key, Class returnType) { - String sql = """ + var sql = """ SELECT status, expiration_time_millis, response FROM %s WHERE key_id = ? AND process_name = ? """.formatted(tableName); try { @@ -67,7 +67,7 @@ public Value mapRow(ResultSet rs, int rowNum) throws SQLException { @Override public void store(IdempotentKey key, Value value) { try { - String responseJson = value.response() != null ? jsonMapper.writeValueAsString(value.response()) : null; + var responseJson = value.response() != null ? jsonMapper.writeValueAsString(value.response()) : null; switch (dialect) { case POSTGRES -> storePostgres(key, value, responseJson); @@ -81,7 +81,7 @@ public void store(IdempotentKey key, Value value) { } private void storePostgres(IdempotentKey key, Value value, String responseJson) { - String sql = """ + var sql = """ INSERT INTO %s (key_id, process_name, status, expiration_time_millis, response) VALUES (?, ?, ?, ?, ?) """.formatted(tableName); @@ -90,7 +90,7 @@ private void storePostgres(IdempotentKey key, Value value, String responseJson) } private void storeMySQL(IdempotentKey key, Value value, String responseJson) { - String sql = """ + var sql = """ INSERT INTO %s (key_id, process_name, status, expiration_time_millis, response) VALUES (?, ?, ?, ?, ?) """.formatted(tableName); jdbcTemplate.update( @@ -98,7 +98,7 @@ private void storeMySQL(IdempotentKey key, Value value, String responseJson) { } private void storeGeneric(IdempotentKey key, Value value, String response) { - String sql = """ + var sql = """ INSERT INTO %s (key_id, process_name, status, expiration_time_millis, response) VALUES (?, ?, ?, ?, ?) """.formatted(tableName); jdbcTemplate.update( @@ -107,7 +107,7 @@ private void storeGeneric(IdempotentKey key, Value value, String response) { @Override public void remove(IdempotentKey key) { - String sql = """ + var sql = """ DELETE FROM %s WHERE key_id = ? AND process_name = ? """.formatted(tableName); jdbcTemplate.update(sql, key.key(), key.processName()); @@ -116,9 +116,9 @@ public void remove(IdempotentKey key) { @Override public void update(IdempotentKey key, Value value) { try { - String responseJson = value.response() != null ? jsonMapper.writeValueAsString(value.response()) : null; + var responseJson = value.response() != null ? jsonMapper.writeValueAsString(value.response()) : null; // Simple UPDATE - trust the caller's contract that this is for INPROGRESS keys - String updateSql = """ + var updateSql = """ UPDATE %s SET status = ?, expiration_time_millis = ?, response = ? WHERE key_id = ? AND process_name = ? """.formatted(tableName); diff --git a/idempotent-rds/src/main/java/io/github/arun0009/idempotent/rds/RdsJacksonJsonBuilderCustomizer.java b/idempotent-rds/src/main/java/io/github/arun0009/idempotent/rds/RdsJacksonJsonBuilderCustomizer.java new file mode 100644 index 0000000..4ee5995 --- /dev/null +++ b/idempotent-rds/src/main/java/io/github/arun0009/idempotent/rds/RdsJacksonJsonBuilderCustomizer.java @@ -0,0 +1,17 @@ +package io.github.arun0009.idempotent.rds; + +import tools.jackson.databind.json.JsonMapper; + +/** + * Customizer for the JsonMapper.Builder used in the RDS idempotent store. + */ +@FunctionalInterface +public interface RdsJacksonJsonBuilderCustomizer { + + /** + * Customizes the provided {@link JsonMapper.Builder} instance. + * + * @param builder the {@link JsonMapper.Builder} instance to customize + */ + void customize(JsonMapper.Builder builder); +} diff --git a/idempotent-rds/src/main/resources/META-INF/spring/org.springframework.boot.autoconfigure.AutoConfiguration.imports b/idempotent-rds/src/main/resources/META-INF/spring/org.springframework.boot.autoconfigure.AutoConfiguration.imports index 36345c8..34544d4 100644 --- a/idempotent-rds/src/main/resources/META-INF/spring/org.springframework.boot.autoconfigure.AutoConfiguration.imports +++ b/idempotent-rds/src/main/resources/META-INF/spring/org.springframework.boot.autoconfigure.AutoConfiguration.imports @@ -1 +1 @@ -io.github.arun0009.idempotent.rds.RdsConfig +io.github.arun0009.idempotent.rds.RdsAutoConfiguration diff --git a/idempotent-rds/src/test/java/io/github/arun0009/idempotent/rds/RdsIdempotentStoreH2Test.java b/idempotent-rds/src/test/java/io/github/arun0009/idempotent/rds/RdsIdempotentStoreH2Test.java index 41f67f2..0dc21db 100644 --- a/idempotent-rds/src/test/java/io/github/arun0009/idempotent/rds/RdsIdempotentStoreH2Test.java +++ b/idempotent-rds/src/test/java/io/github/arun0009/idempotent/rds/RdsIdempotentStoreH2Test.java @@ -1,5 +1,6 @@ package io.github.arun0009.idempotent.rds; +import com.zaxxer.hikari.HikariDataSource; import io.github.arun0009.idempotent.core.persistence.IdempotentStore; import io.github.arun0009.idempotent.core.persistence.IdempotentStore.IdempotentKey; import io.github.arun0009.idempotent.core.persistence.IdempotentStore.Value; @@ -7,7 +8,6 @@ import org.junit.jupiter.api.Test; import org.junit.jupiter.api.extension.ExtendWith; import org.springframework.beans.factory.annotation.Autowired; -import org.springframework.boot.jdbc.DataSourceBuilder; import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration; import org.springframework.dao.DuplicateKeyException; @@ -46,12 +46,12 @@ public class RdsIdempotentStoreH2Test { static class H2TestConfig { @Bean public DataSource dataSource() { - return DataSourceBuilder.create() - .url("jdbc:h2:mem:testdb;DB_CLOSE_DELAY=-1;MODE=MySQL") - .driverClassName("org.h2.Driver") - .username("sa") - .password("") - .build(); + HikariDataSource ds = new HikariDataSource(); + ds.setJdbcUrl("jdbc:h2:mem:testdb;DB_CLOSE_DELAY=-1;MODE=MySQL"); + ds.setDriverClassName("org.h2.Driver"); + ds.setUsername("sa"); + ds.setPassword(""); + return ds; } @Bean diff --git a/idempotent-rds/src/test/java/io/github/arun0009/idempotent/rds/RdsIdempotentStoreMySQLTest.java b/idempotent-rds/src/test/java/io/github/arun0009/idempotent/rds/RdsIdempotentStoreMySQLTest.java index c60a7da..a40294e 100644 --- a/idempotent-rds/src/test/java/io/github/arun0009/idempotent/rds/RdsIdempotentStoreMySQLTest.java +++ b/idempotent-rds/src/test/java/io/github/arun0009/idempotent/rds/RdsIdempotentStoreMySQLTest.java @@ -70,7 +70,9 @@ public void testStoreAndGet() { Value retrieved = idempotentStore.getValue(key, Map.class); assertNotNull(retrieved); assertEquals("COMPLETED", retrieved.status()); - assertEquals("success", ((Map) retrieved.response()).get("result")); + @SuppressWarnings("unchecked") + Map response = (Map) retrieved.response(); + assertEquals("success", response.get("result")); } @Test diff --git a/idempotent-rds/src/test/java/io/github/arun0009/idempotent/rds/RdsIdempotentStorePostgreSQLTest.java b/idempotent-rds/src/test/java/io/github/arun0009/idempotent/rds/RdsIdempotentStorePostgreSQLTest.java index 16996e9..16259a4 100644 --- a/idempotent-rds/src/test/java/io/github/arun0009/idempotent/rds/RdsIdempotentStorePostgreSQLTest.java +++ b/idempotent-rds/src/test/java/io/github/arun0009/idempotent/rds/RdsIdempotentStorePostgreSQLTest.java @@ -65,7 +65,9 @@ public void testStoreAndGet() { Value retrieved = idempotentStore.getValue(key, Map.class); assertNotNull(retrieved); assertEquals("COMPLETED", retrieved.status()); - assertEquals("success", ((Map) retrieved.response()).get("result")); + @SuppressWarnings("unchecked") + Map response = (Map) retrieved.response(); + assertEquals("success", response.get("result")); } @Test diff --git a/idempotent-rds/src/test/java/io/github/arun0009/idempotent/rds/RdsMySQLTestConfig.java b/idempotent-rds/src/test/java/io/github/arun0009/idempotent/rds/RdsMySQLTestConfig.java index 7c1d527..9092411 100644 --- a/idempotent-rds/src/test/java/io/github/arun0009/idempotent/rds/RdsMySQLTestConfig.java +++ b/idempotent-rds/src/test/java/io/github/arun0009/idempotent/rds/RdsMySQLTestConfig.java @@ -1,7 +1,7 @@ package io.github.arun0009.idempotent.rds; +import com.zaxxer.hikari.HikariDataSource; import io.github.arun0009.idempotent.core.persistence.IdempotentStore; -import org.springframework.boot.jdbc.DataSourceBuilder; import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration; import org.springframework.jdbc.core.JdbcTemplate; @@ -32,12 +32,12 @@ public class RdsMySQLTestConfig { @Bean public DataSource dataSource() { - return DataSourceBuilder.create() - .url(mysql.getJdbcUrl()) - .driverClassName(mysql.getDriverClassName()) - .username(mysql.getUsername()) - .password(mysql.getPassword()) - .build(); + HikariDataSource ds = new HikariDataSource(); + ds.setJdbcUrl(mysql.getJdbcUrl()); + ds.setDriverClassName(mysql.getDriverClassName()); + ds.setUsername(mysql.getUsername()); + ds.setPassword(mysql.getPassword()); + return ds; } @Bean diff --git a/idempotent-rds/src/test/java/io/github/arun0009/idempotent/rds/RdsPostgreSQLTestConfig.java b/idempotent-rds/src/test/java/io/github/arun0009/idempotent/rds/RdsPostgreSQLTestConfig.java index 22eec9e..5610095 100644 --- a/idempotent-rds/src/test/java/io/github/arun0009/idempotent/rds/RdsPostgreSQLTestConfig.java +++ b/idempotent-rds/src/test/java/io/github/arun0009/idempotent/rds/RdsPostgreSQLTestConfig.java @@ -1,7 +1,7 @@ package io.github.arun0009.idempotent.rds; +import com.zaxxer.hikari.HikariDataSource; import io.github.arun0009.idempotent.core.persistence.IdempotentStore; -import org.springframework.boot.jdbc.DataSourceBuilder; import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration; import org.springframework.jdbc.core.JdbcTemplate; @@ -32,12 +32,12 @@ public class RdsPostgreSQLTestConfig { @Bean public DataSource dataSource() { - return DataSourceBuilder.create() - .url(postgres.getJdbcUrl()) - .driverClassName(postgres.getDriverClassName()) - .username(postgres.getUsername()) - .password(postgres.getPassword()) - .build(); + HikariDataSource ds = new HikariDataSource(); + ds.setJdbcUrl(postgres.getJdbcUrl()); + ds.setDriverClassName(postgres.getDriverClassName()); + ds.setUsername(postgres.getUsername()); + ds.setPassword(postgres.getPassword()); + return ds; } @Bean diff --git a/pom.xml b/pom.xml index 3437b1e..877a833 100644 --- a/pom.xml +++ b/pom.xml @@ -49,6 +49,7 @@ ${java.version} 3.2.1 4.0.3 + 1.21.4 @@ -84,6 +85,12 @@ 3.27.7 test + + org.testcontainers + junit-jupiter + ${testcontainers.version} + test + @@ -121,24 +128,6 @@ testcontainers test - - org.testcontainers - junit-jupiter - ${testcontainers.version} - test - - - org.testcontainers - mysql - ${testcontainers.version} - test - - - org.testcontainers - postgresql - ${testcontainers.version} - test - From 127e270bab0725a1e8a452c625b6a2947307bd19 Mon Sep 17 00:00:00 2001 From: Arun Gopalpuri Date: Tue, 24 Feb 2026 07:26:24 -0800 Subject: [PATCH 6/6] add javadocs --- .../rds/RdsIdempotentProperties.java | 32 +++++++++++++++---- 1 file changed, 26 insertions(+), 6 deletions(-) diff --git a/idempotent-rds/src/main/java/io/github/arun0009/idempotent/rds/RdsIdempotentProperties.java b/idempotent-rds/src/main/java/io/github/arun0009/idempotent/rds/RdsIdempotentProperties.java index cba70fe..44b0677 100644 --- a/idempotent-rds/src/main/java/io/github/arun0009/idempotent/rds/RdsIdempotentProperties.java +++ b/idempotent-rds/src/main/java/io/github/arun0009/idempotent/rds/RdsIdempotentProperties.java @@ -3,18 +3,38 @@ import org.springframework.boot.context.properties.ConfigurationProperties; import org.springframework.boot.context.properties.bind.DefaultValue; +/** + * Configuration properties for RDS-based idempotency implementation. + *

+ * These properties control the behavior of the idempotent key storage and + * cleanup + * in a relational database system. + *

+ * + * @param tableName the name of the database table used to store idempotent keys + * @param cleanup configuration for the cleanup task that removes expired + * idempotent records + */ @ConfigurationProperties(prefix = "idempotent.rds") public record RdsIdempotentProperties( - /** Name of the database table for idempotent keys. */ @DefaultValue("idempotent") String tableName, Cleanup cleanup) { + /** + * Configuration properties for the cleanup task that removes expired idempotent + * records. + *

+ * The cleanup task runs periodically to delete expired idempotent keys from the + * database, preventing unbounded growth of the idempotent key table. + *

+ * + * @param enabled whether the cleanup task should run + * @param batchSize maximum number of expired records to delete in a single + * cleanup operation + * @param fixedDelay delay in milliseconds between consecutive cleanup task + * executions + */ public record Cleanup( - /** Whether the cleanup task is enabled. Set to false for CDS or AOT cache runs. */ @DefaultValue("true") boolean enabled, - - /** Number of expired records to delete per batch. */ @DefaultValue("1000") int batchSize, - - /** Fixed delay in milliseconds between cleanup runs. */ @DefaultValue("60000") long fixedDelay) {} }