This is an automated email from the ASF dual-hosted git repository.
gnodet pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/camel.git
The following commit(s) were added to refs/heads/main by this push:
new dbb0045881bc CAMEL-24463: Add CAS methods and Duration TTL to
KeyValueRepository SPI (#25863)
dbb0045881bc is described below
commit dbb0045881bcbfc8b422dddaa061fdd98e8906d0
Author: Guillaume Nodet <[email protected]>
AuthorDate: Tue Sep 1 11:38:23 2026 +0200
CAMEL-24463: Add CAS methods and Duration TTL to KeyValueRepository SPI
(#25863)
Add compare-and-swap operations to KeyValueRepository:
- replace(key, expectedOldValue, newValue, ttl): atomic CAS replace
- delete(key, expectedValue): atomic CAS delete
Change TTL parameter from long millis to java.time.Duration across the
entire SPI for type safety. null/zero/negative Duration means no expiry.
Updated MemoryKeyValueRepository, KeyValueAggregationRepository,
KeyValueIdempotentRepository, and StateStoreProducer accordingly.
Co-authored-by: Claude Opus 4.6 <[email protected]>
---
.../component/statestore/StateStoreProducer.java | 4 +-
.../org/apache/camel/spi/KeyValueRepository.java | 72 +++++++++--
.../support/KeyValueAggregationRepository.java | 4 +-
.../support/KeyValueIdempotentRepository.java | 2 +-
.../camel/support/MemoryKeyValueRepository.java | 44 ++++++-
.../support/KeyValueIdempotentRepositoryTest.java | 4 +-
.../support/MemoryKeyValueRepositoryTest.java | 131 ++++++++++++++++-----
7 files changed, 211 insertions(+), 50 deletions(-)
diff --git
a/components/camel-state-store/camel-state-store/src/main/java/org/apache/camel/component/statestore/StateStoreProducer.java
b/components/camel-state-store/camel-state-store/src/main/java/org/apache/camel/component/statestore/StateStoreProducer.java
index 4f4a175c6b11..2cb03af7218e 100644
---
a/components/camel-state-store/camel-state-store/src/main/java/org/apache/camel/component/statestore/StateStoreProducer.java
+++
b/components/camel-state-store/camel-state-store/src/main/java/org/apache/camel/component/statestore/StateStoreProducer.java
@@ -16,6 +16,7 @@
*/
package org.apache.camel.component.statestore;
+import java.time.Duration;
import java.util.Arrays;
import org.apache.camel.Exchange;
@@ -46,7 +47,8 @@ public class StateStoreProducer extends DefaultProducer {
}
KeyValueRepository backend = endpoint.getBackend();
- long ttl = determineTtl(exchange);
+ long ttlMillis = determineTtl(exchange);
+ Duration ttl = ttlMillis > 0 ? Duration.ofMillis(ttlMillis) : null;
Message message = exchange.getMessage();
switch (op) {
diff --git
a/core/camel-api/src/main/java/org/apache/camel/spi/KeyValueRepository.java
b/core/camel-api/src/main/java/org/apache/camel/spi/KeyValueRepository.java
index cd7d6f8a1ac6..fef877edda4c 100644
--- a/core/camel-api/src/main/java/org/apache/camel/spi/KeyValueRepository.java
+++ b/core/camel-api/src/main/java/org/apache/camel/spi/KeyValueRepository.java
@@ -16,6 +16,8 @@
*/
package org.apache.camel.spi;
+import java.time.Duration;
+import java.util.Objects;
import java.util.Set;
import org.apache.camel.Service;
@@ -33,8 +35,8 @@ import org.jspecify.annotations.Nullable;
* storage technology (Redis, Hazelcast, Infinispan, JDBC, etc.), a single
{@code KeyValueRepository} implementation can
* be wrapped by the appropriate adapter.
* <p/>
- * Implementations must be thread-safe. Entries may optionally have a
time-to-live (TTL); a TTL of {@code 0} or less
- * means the entry does not expire.
+ * Implementations must be thread-safe. Entries may optionally have a
time-to-live (TTL); a {@code null}, zero, or
+ * negative TTL means the entry does not expire.
*
* @since 4.23
*/
@@ -52,13 +54,13 @@ public interface KeyValueRepository extends Service {
/**
* Stores a value under the given key with an optional time-to-live.
*
- * @param key the key
- * @param value the value to store
- * @param ttlMillis the time-to-live in milliseconds; {@code 0} or
negative means no expiration
- * @return the previous value associated with the key, or {@code
null} if there was no mapping
+ * @param key the key
+ * @param value the value to store
+ * @param ttl the time-to-live; {@code null}, zero, or negative means
no expiration
+ * @return the previous value associated with the key, or {@code
null} if there was no mapping
*/
@Nullable
- Object put(String key, Object value, long ttlMillis);
+ Object put(String key, Object value, @Nullable Duration ttl);
/**
* Removes the entry for the given key.
@@ -95,21 +97,65 @@ public interface KeyValueRepository extends Service {
* The default implementation is not atomic. Implementations backed by
stores that support atomic compare-and-set
* operations should override this method for better concurrency
guarantees.
*
- * @param key the key
- * @param value the value to store
- * @param ttlMillis the time-to-live in milliseconds; {@code 0} or
negative means no expiration
- * @return the existing value if the key was already present, or
{@code null} if the put succeeded
+ * @param key the key
+ * @param value the value to store
+ * @param ttl the time-to-live; {@code null}, zero, or negative means
no expiration
+ * @return the existing value if the key was already present, or
{@code null} if the put succeeded
*/
@Nullable
- default Object putIfAbsent(String key, Object value, long ttlMillis) {
+ default Object putIfAbsent(String key, Object value, @Nullable Duration
ttl) {
Object existing = get(key);
if (existing != null) {
return existing;
}
- put(key, value, ttlMillis);
+ put(key, value, ttl);
return null;
}
+ /**
+ * Atomically replaces the value for the given key only if the current
value equals the expected old value
+ * (compare-and-swap).
+ * <p/>
+ * The default implementation is not atomic. Implementations backed by
stores that support atomic compare-and-swap
+ * operations (e.g., {@code ConcurrentMap.replace}, Hazelcast {@code
IMap.replace}) should override this method for
+ * better concurrency guarantees.
+ *
+ * @param key the key
+ * @param expectedOldValue the value that must currently be associated
with the key
+ * @param newValue the new value to store
+ * @param ttl the time-to-live for the new entry; {@code
null}, zero, or negative means no expiration
+ * @return {@code true} if the value was replaced, {@code
false} if the current value did not match
+ */
+ default boolean replace(String key, Object expectedOldValue, Object
newValue, @Nullable Duration ttl) {
+ Object current = get(key);
+ if (current != null && Objects.equals(current, expectedOldValue)) {
+ put(key, newValue, ttl);
+ return true;
+ }
+ return false;
+ }
+
+ /**
+ * Removes the entry for the given key only if the current value equals
the expected value (compare-and-swap).
+ * <p/>
+ * The default implementation is not atomic. Implementations backed by
stores that support atomic compare-and-remove
+ * operations (e.g., {@code ConcurrentMap.remove(key, value)}, Hazelcast
{@code IMap.remove(key, value)}) should
+ * override this method for better concurrency guarantees.
+ *
+ * @param key the key to remove
+ * @param expectedValue the value that must currently be associated with
the key
+ * @return {@code true} if the entry was removed, {@code
false} if the current value did not match or
+ * the key was not present
+ */
+ default boolean delete(String key, Object expectedValue) {
+ Object current = get(key);
+ if (current != null && Objects.equals(current, expectedValue)) {
+ delete(key);
+ return true;
+ }
+ return false;
+ }
+
/**
* Returns the number of non-expired entries in the repository.
*
diff --git
a/core/camel-support/src/main/java/org/apache/camel/support/KeyValueAggregationRepository.java
b/core/camel-support/src/main/java/org/apache/camel/support/KeyValueAggregationRepository.java
index bfa46ae46514..95fb22141b5b 100644
---
a/core/camel-support/src/main/java/org/apache/camel/support/KeyValueAggregationRepository.java
+++
b/core/camel-support/src/main/java/org/apache/camel/support/KeyValueAggregationRepository.java
@@ -110,7 +110,7 @@ public class KeyValueAggregationRepository extends
ServiceSupport
public Exchange add(CamelContext camelContext, String key, Exchange
exchange) {
LOG.trace("Adding an Exchange with ID {} for key {}",
exchange.getExchangeId(), key);
DefaultExchangeHolder newHolder =
DefaultExchangeHolder.marshal(exchange, true, allowSerializedHeaders);
- DefaultExchangeHolder oldHolder = (DefaultExchangeHolder)
repository.put(AGGREGATE_PREFIX + key, newHolder, 0);
+ DefaultExchangeHolder oldHolder = (DefaultExchangeHolder)
repository.put(AGGREGATE_PREFIX + key, newHolder, null);
return unmarshallExchange(camelContext, oldHolder);
}
@@ -126,7 +126,7 @@ public class KeyValueAggregationRepository extends
ServiceSupport
if (useRecovery && holder != null) {
// Store under the exchangeId for potential recovery
LOG.trace("Moving Exchange with ID {} to completed (pending
confirmation)", exchange.getExchangeId());
- repository.put(COMPLETED_PREFIX + exchange.getExchangeId(),
holder, 0);
+ repository.put(COMPLETED_PREFIX + exchange.getExchangeId(),
holder, null);
}
}
diff --git
a/core/camel-support/src/main/java/org/apache/camel/support/KeyValueIdempotentRepository.java
b/core/camel-support/src/main/java/org/apache/camel/support/KeyValueIdempotentRepository.java
index 010e913361a2..d11b656641e1 100644
---
a/core/camel-support/src/main/java/org/apache/camel/support/KeyValueIdempotentRepository.java
+++
b/core/camel-support/src/main/java/org/apache/camel/support/KeyValueIdempotentRepository.java
@@ -84,7 +84,7 @@ public class KeyValueIdempotentRepository extends
ServiceSupport implements Idem
@Override
public boolean add(String key) {
// putIfAbsent returns null if the key was successfully added (not
already present)
- return repository.putIfAbsent(IDEMPOTENT_PREFIX + key, Boolean.TRUE,
0) == null;
+ return repository.putIfAbsent(IDEMPOTENT_PREFIX + key, Boolean.TRUE,
null) == null;
}
@Override
diff --git
a/core/camel-support/src/main/java/org/apache/camel/support/MemoryKeyValueRepository.java
b/core/camel-support/src/main/java/org/apache/camel/support/MemoryKeyValueRepository.java
index f32527ea9094..d6ce48680ca9 100644
---
a/core/camel-support/src/main/java/org/apache/camel/support/MemoryKeyValueRepository.java
+++
b/core/camel-support/src/main/java/org/apache/camel/support/MemoryKeyValueRepository.java
@@ -18,8 +18,10 @@ package org.apache.camel.support;
import java.io.Serial;
import java.io.Serializable;
+import java.time.Duration;
import java.util.Iterator;
import java.util.Map;
+import java.util.Objects;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import java.util.stream.Collectors;
@@ -73,8 +75,8 @@ public class MemoryKeyValueRepository extends ServiceSupport
implements KeyValue
@Override
@ManagedOperation(description = "Put a key-value pair with optional TTL")
- public Object put(String key, Object value, long ttlMillis) {
- long expiresAt = ttlMillis > 0 ? System.currentTimeMillis() +
ttlMillis : Long.MAX_VALUE;
+ public Object put(String key, Object value, Duration ttl) {
+ long expiresAt = toExpiresAt(ttl);
Entry previous = store.put(key, new Entry(value, expiresAt));
if (previous == null) {
return null;
@@ -130,8 +132,8 @@ public class MemoryKeyValueRepository extends
ServiceSupport implements KeyValue
}
@Override
- public Object putIfAbsent(String key, Object value, long ttlMillis) {
- long expiresAt = ttlMillis > 0 ? System.currentTimeMillis() +
ttlMillis : Long.MAX_VALUE;
+ public Object putIfAbsent(String key, Object value, Duration ttl) {
+ long expiresAt = toExpiresAt(ttl);
Entry newEntry = new Entry(value, expiresAt);
Entry existing = store.putIfAbsent(key, newEntry);
if (existing == null) {
@@ -149,6 +151,33 @@ public class MemoryKeyValueRepository extends
ServiceSupport implements KeyValue
return existing.value();
}
+ @Override
+ public boolean replace(String key, Object expectedOldValue, Object
newValue, Duration ttl) {
+ long expiresAt = toExpiresAt(ttl);
+ boolean[] replaced = { false };
+ store.computeIfPresent(key, (k, current) -> {
+ if (!current.isExpired() && Objects.equals(current.value(),
expectedOldValue)) {
+ replaced[0] = true;
+ return new Entry(newValue, expiresAt);
+ }
+ return current;
+ });
+ return replaced[0];
+ }
+
+ @Override
+ public boolean delete(String key, Object expectedValue) {
+ boolean[] removed = { false };
+ store.computeIfPresent(key, (k, current) -> {
+ if (!current.isExpired() && Objects.equals(current.value(),
expectedValue)) {
+ removed[0] = true;
+ return null; // returning null removes the entry from the map
+ }
+ return current;
+ });
+ return removed[0];
+ }
+
@Override
@ManagedAttribute(description = "The number of entries in the repository")
public int size() {
@@ -161,6 +190,13 @@ public class MemoryKeyValueRepository extends
ServiceSupport implements KeyValue
store.clear();
}
+ private static long toExpiresAt(Duration ttl) {
+ if (ttl == null || ttl.isZero() || ttl.isNegative()) {
+ return Long.MAX_VALUE;
+ }
+ return System.currentTimeMillis() + ttl.toMillis();
+ }
+
private void evictExpired() {
Iterator<Map.Entry<String, Entry>> it = store.entrySet().iterator();
while (it.hasNext()) {
diff --git
a/core/camel-support/src/test/java/org/apache/camel/support/KeyValueIdempotentRepositoryTest.java
b/core/camel-support/src/test/java/org/apache/camel/support/KeyValueIdempotentRepositoryTest.java
index 6148fb5444ac..c1853e4c172a 100644
---
a/core/camel-support/src/test/java/org/apache/camel/support/KeyValueIdempotentRepositoryTest.java
+++
b/core/camel-support/src/test/java/org/apache/camel/support/KeyValueIdempotentRepositoryTest.java
@@ -145,7 +145,7 @@ class KeyValueIdempotentRepositoryTest {
@Test
void testClearDoesNotAffectOtherPrefixes() {
// Simulate another adapter storing entries under a different prefix
- kvRepository.put("aggregate:order-1", "exchange-holder", 0);
+ kvRepository.put("aggregate:order-1", "exchange-holder", null);
// Add idempotent entries and clear them
idempotentRepository.add("msg-001");
@@ -166,7 +166,7 @@ class KeyValueIdempotentRepositoryTest {
idempotentRepository.add("order-1");
// A different adapter storing under its own prefix should not collide
- kvRepository.put("aggregate:order-1", "exchange-data", 0);
+ kvRepository.put("aggregate:order-1", "exchange-data", null);
// The idempotent entry should still resolve correctly
assertThat(idempotentRepository.contains("order-1")).isTrue();
diff --git
a/core/camel-support/src/test/java/org/apache/camel/support/MemoryKeyValueRepositoryTest.java
b/core/camel-support/src/test/java/org/apache/camel/support/MemoryKeyValueRepositoryTest.java
index f8f9465cb255..e72a8f71ae26 100644
---
a/core/camel-support/src/test/java/org/apache/camel/support/MemoryKeyValueRepositoryTest.java
+++
b/core/camel-support/src/test/java/org/apache/camel/support/MemoryKeyValueRepositoryTest.java
@@ -16,6 +16,7 @@
*/
package org.apache.camel.support;
+import java.time.Duration;
import java.util.Set;
import java.util.concurrent.TimeUnit;
@@ -43,7 +44,7 @@ class MemoryKeyValueRepositoryTest {
@Test
void testPutAndGet() {
- repository.put("key1", "value1", 0);
+ repository.put("key1", "value1", null);
assertThat(repository.get("key1")).isEqualTo("value1");
}
@@ -55,15 +56,15 @@ class MemoryKeyValueRepositoryTest {
@Test
void testPutOverwritesExistingValue() {
- repository.put("key1", "value1", 0);
- repository.put("key1", "value2", 0);
+ repository.put("key1", "value1", null);
+ repository.put("key1", "value2", null);
assertThat(repository.get("key1")).isEqualTo("value2");
}
@Test
void testDelete() {
- repository.put("key1", "value1", 0);
+ repository.put("key1", "value1", null);
Object deleted = repository.delete("key1");
@@ -78,7 +79,7 @@ class MemoryKeyValueRepositoryTest {
@Test
void testContains() {
- repository.put("key1", "value1", 0);
+ repository.put("key1", "value1", null);
assertThat(repository.contains("key1")).isTrue();
assertThat(repository.contains("nonexistent")).isFalse();
@@ -86,9 +87,9 @@ class MemoryKeyValueRepositoryTest {
@Test
void testKeys() {
- repository.put("key1", "value1", 0);
- repository.put("key2", "value2", 0);
- repository.put("key3", "value3", 0);
+ repository.put("key1", "value1", null);
+ repository.put("key2", "value2", null);
+ repository.put("key3", "value3", null);
Set<String> keys = repository.keys();
@@ -102,8 +103,8 @@ class MemoryKeyValueRepositoryTest {
@Test
void testClear() {
- repository.put("key1", "value1", 0);
- repository.put("key2", "value2", 0);
+ repository.put("key1", "value1", null);
+ repository.put("key2", "value2", null);
repository.clear();
@@ -116,10 +117,10 @@ class MemoryKeyValueRepositoryTest {
void testSize() {
assertThat(repository.size()).isZero();
- repository.put("key1", "value1", 0);
+ repository.put("key1", "value1", null);
assertThat(repository.size()).isEqualTo(1);
- repository.put("key2", "value2", 0);
+ repository.put("key2", "value2", null);
assertThat(repository.size()).isEqualTo(2);
repository.delete("key1");
@@ -128,7 +129,7 @@ class MemoryKeyValueRepositoryTest {
@Test
void testPutIfAbsentNewKey() {
- Object result = repository.putIfAbsent("key1", "value1", 0);
+ Object result = repository.putIfAbsent("key1", "value1", null);
assertThat(result).isNull();
assertThat(repository.get("key1")).isEqualTo("value1");
@@ -136,9 +137,9 @@ class MemoryKeyValueRepositoryTest {
@Test
void testPutIfAbsentExistingKey() {
- repository.put("key1", "value1", 0);
+ repository.put("key1", "value1", null);
- Object result = repository.putIfAbsent("key1", "value2", 0);
+ Object result = repository.putIfAbsent("key1", "value2", null);
assertThat(result).isEqualTo("value1");
assertThat(repository.get("key1")).isEqualTo("value1");
@@ -147,7 +148,7 @@ class MemoryKeyValueRepositoryTest {
@Test
void testTtlExpiration() {
// Use a very short TTL
- repository.put("key1", "value1", 50);
+ repository.put("key1", "value1", Duration.ofMillis(50));
assertThat(repository.get("key1")).isEqualTo("value1");
assertThat(repository.contains("key1")).isTrue();
@@ -162,8 +163,8 @@ class MemoryKeyValueRepositoryTest {
@Test
void testTtlExpirationOnKeys() {
- repository.put("key1", "value1", 50);
- repository.put("key2", "value2", 0); // no expiration
+ repository.put("key1", "value1", Duration.ofMillis(50));
+ repository.put("key2", "value2", null); // no expiration
await().atMost(500, TimeUnit.MILLISECONDS)
.untilAsserted(() -> {
@@ -174,7 +175,7 @@ class MemoryKeyValueRepositoryTest {
@Test
void testTtlExpirationOnDelete() {
- repository.put("key1", "value1", 50);
+ repository.put("key1", "value1", Duration.ofMillis(50));
await().atMost(500, TimeUnit.MILLISECONDS)
.untilAsserted(() -> {
@@ -185,29 +186,38 @@ class MemoryKeyValueRepositoryTest {
@Test
void testPutIfAbsentWithExpiredEntry() {
- repository.put("key1", "value1", 50);
+ repository.put("key1", "value1", Duration.ofMillis(50));
await().atMost(500, TimeUnit.MILLISECONDS)
.untilAsserted(() -> {
// The entry has expired, so putIfAbsent should succeed
- Object result = repository.putIfAbsent("key1", "value2",
0);
+ Object result = repository.putIfAbsent("key1", "value2",
null);
assertThat(result).isNull();
assertThat(repository.get("key1")).isEqualTo("value2");
});
}
+ @Test
+ void testNoTtlWithNull() {
+ repository.put("key1", "value1", null);
+
+ // Entry with null TTL should not expire
+ assertThat(repository.get("key1")).isEqualTo("value1");
+ assertThat(repository.contains("key1")).isTrue();
+ }
+
@Test
void testNoTtlWithZero() {
- repository.put("key1", "value1", 0);
+ repository.put("key1", "value1", Duration.ZERO);
- // Entry with TTL=0 should not expire
+ // Entry with zero TTL should not expire
assertThat(repository.get("key1")).isEqualTo("value1");
assertThat(repository.contains("key1")).isTrue();
}
@Test
void testNoTtlWithNegative() {
- repository.put("key1", "value1", -1);
+ repository.put("key1", "value1", Duration.ofMillis(-1));
// Entry with negative TTL should not expire
assertThat(repository.get("key1")).isEqualTo("value1");
@@ -216,12 +226,79 @@ class MemoryKeyValueRepositoryTest {
@Test
void testStoresDifferentValueTypes() {
- repository.put("string", "hello", 0);
- repository.put("integer", 42, 0);
- repository.put("boolean", Boolean.TRUE, 0);
+ repository.put("string", "hello", null);
+ repository.put("integer", 42, null);
+ repository.put("boolean", Boolean.TRUE, null);
assertThat(repository.get("string")).isEqualTo("hello");
assertThat(repository.get("integer")).isEqualTo(42);
assertThat(repository.get("boolean")).isEqualTo(Boolean.TRUE);
}
+
+ @Test
+ void testReplaceMatchingValue() {
+ repository.put("key1", "value1", null);
+
+ boolean replaced = repository.replace("key1", "value1", "value2",
null);
+
+ assertThat(replaced).isTrue();
+ assertThat(repository.get("key1")).isEqualTo("value2");
+ }
+
+ @Test
+ void testReplaceNonMatchingValue() {
+ repository.put("key1", "value1", null);
+
+ boolean replaced = repository.replace("key1", "wrong", "value2", null);
+
+ assertThat(replaced).isFalse();
+ assertThat(repository.get("key1")).isEqualTo("value1");
+ }
+
+ @Test
+ void testReplaceMissingKey() {
+ boolean replaced = repository.replace("nonexistent", "value1",
"value2", null);
+
+ assertThat(replaced).isFalse();
+ }
+
+ @Test
+ void testReplaceWithTtl() {
+ repository.put("key1", "value1", null);
+
+ boolean replaced = repository.replace("key1", "value1", "value2",
Duration.ofMillis(500));
+
+ assertThat(replaced).isTrue();
+ assertThat(repository.get("key1")).isEqualTo("value2");
+
+ await().atMost(5, TimeUnit.SECONDS)
+ .untilAsserted(() ->
assertThat(repository.get("key1")).isNull());
+ }
+
+ @Test
+ void testDeleteWithMatchingValue() {
+ repository.put("key1", "value1", null);
+
+ boolean deleted = repository.delete("key1", "value1");
+
+ assertThat(deleted).isTrue();
+ assertThat(repository.get("key1")).isNull();
+ }
+
+ @Test
+ void testDeleteWithNonMatchingValue() {
+ repository.put("key1", "value1", null);
+
+ boolean deleted = repository.delete("key1", "wrong");
+
+ assertThat(deleted).isFalse();
+ assertThat(repository.get("key1")).isEqualTo("value1");
+ }
+
+ @Test
+ void testDeleteWithMissingKey() {
+ boolean deleted = repository.delete("nonexistent", "value1");
+
+ assertThat(deleted).isFalse();
+ }
}
