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 c01f4c648e30 CAMEL-24463: Improve KeyValueRepository SPI
c01f4c648e30 is described below
commit c01f4c648e30b38dd94e81ea10bb3334f117a110
Author: Guillaume Nodet <[email protected]>
AuthorDate: Wed Sep 2 06:48:29 2026 +0200
CAMEL-24463: Improve KeyValueRepository SPI
- Add put(key, value) and putIfAbsent(key, value) convenience defaults
that delegate to their Duration-based variants with null TTL
- Add KeyValueRepositoryHelper with shared serialization utilities
(serialize/deserialize objects to byte arrays) for persistent backends
- Update KeyValueAggregationRepository and KeyValueIdempotentRepository
to use the new convenience methods
Co-authored-by: Claude Opus 4.6 <[email protected]>
---
.../org/apache/camel/spi/KeyValueRepository.java | 28 +++++
.../support/KeyValueAggregationRepository.java | 4 +-
.../support/KeyValueIdempotentRepository.java | 2 +-
.../camel/support/KeyValueRepositoryHelper.java | 125 +++++++++++++++++++++
.../support/KeyValueIdempotentRepositoryTest.java | 4 +-
5 files changed, 158 insertions(+), 5 deletions(-)
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 fef877edda4c..7e92d11818ca 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
@@ -51,6 +51,20 @@ public interface KeyValueRepository extends Service {
@Nullable
Object get(String key);
+ /**
+ * Stores a value under the given key with no expiration.
+ * <p/>
+ * Equivalent to {@code put(key, value, null)}.
+ *
+ * @param key the key
+ * @param value the value to store
+ * @return the previous value associated with the key, or {@code
null} if there was no mapping
+ */
+ @Nullable
+ default Object put(String key, Object value) {
+ return put(key, value, null);
+ }
+
/**
* Stores a value under the given key with an optional time-to-live.
*
@@ -91,6 +105,20 @@ public interface KeyValueRepository extends Service {
*/
void clear();
+ /**
+ * Stores the value under the given key only if no non-expired mapping
already exists, with no expiration.
+ * <p/>
+ * Equivalent to {@code putIfAbsent(key, value, null)}.
+ *
+ * @param key the key
+ * @param value the value to store
+ * @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) {
+ return putIfAbsent(key, value, null);
+ }
+
/**
* Stores the value under the given key only if no non-expired mapping
already exists.
* <p/>
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 95fb22141b5b..38a4aaf24301 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, null);
+ DefaultExchangeHolder oldHolder = (DefaultExchangeHolder)
repository.put(AGGREGATE_PREFIX + key, newHolder);
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, null);
+ repository.put(COMPLETED_PREFIX + exchange.getExchangeId(),
holder);
}
}
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 d11b656641e1..02d87434d725 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,
null) == null;
+ return repository.putIfAbsent(IDEMPOTENT_PREFIX + key, Boolean.TRUE)
== null;
}
@Override
diff --git
a/core/camel-support/src/main/java/org/apache/camel/support/KeyValueRepositoryHelper.java
b/core/camel-support/src/main/java/org/apache/camel/support/KeyValueRepositoryHelper.java
new file mode 100644
index 000000000000..13d3ff696456
--- /dev/null
+++
b/core/camel-support/src/main/java/org/apache/camel/support/KeyValueRepositoryHelper.java
@@ -0,0 +1,125 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.camel.support;
+
+import java.io.ByteArrayInputStream;
+import java.io.ByteArrayOutputStream;
+import java.io.IOException;
+import java.io.ObjectInputStream;
+import java.io.ObjectOutputStream;
+import java.nio.ByteBuffer;
+
+import org.apache.camel.RuntimeCamelException;
+
+/**
+ * Shared serialization utilities for {@link
org.apache.camel.spi.KeyValueRepository} implementations.
+ * <p/>
+ * All persistent {@code KeyValueRepository} implementations need to serialize
arbitrary Java objects to bytes (for BLOB
+ * columns, Kafka messages, etc.) and deserialize them back. This helper
centralises that logic to avoid the same
+ * try/catch boilerplate in every implementation.
+ * <p/>
+ * <b>Security note:</b> These methods use plain Java serialization
+ * ({@link ObjectOutputStream}/{@link ObjectInputStream}). The stored data is
trusted — it was written by the same
+ * application instance or cluster. Do not expose a repository's raw byte
store to untrusted input.
+ *
+ * @since 4.23
+ */
+public final class KeyValueRepositoryHelper {
+
+ private KeyValueRepositoryHelper() {
+ // utility class
+ }
+
+ /**
+ * Serializes an object to a byte array using Java object serialization.
+ *
+ * @param value the object to serialize (must be {@link
java.io.Serializable})
+ * @return the serialized bytes
+ * @throws RuntimeCamelException if serialization fails
+ */
+ public static byte[] serialize(Object value) {
+ try (ByteArrayOutputStream bos = new ByteArrayOutputStream();
+ ObjectOutputStream oos = new ObjectOutputStream(bos)) {
+ oos.writeObject(value);
+ oos.flush();
+ return bos.toByteArray();
+ } catch (IOException e) {
+ throw new RuntimeCamelException("Failed to serialize value", e);
+ }
+ }
+
+ /**
+ * Serializes an object to a {@link ByteBuffer} using Java object
serialization. Useful for drivers that work with
+ * {@code ByteBuffer} (e.g. Cassandra).
+ *
+ * @param value the object to serialize (must be {@link
java.io.Serializable})
+ * @return a ByteBuffer wrapping the serialized bytes
+ * @throws RuntimeCamelException if serialization fails
+ */
+ public static ByteBuffer serializeToByteBuffer(Object value) {
+ return ByteBuffer.wrap(serialize(value));
+ }
+
+ /**
+ * Deserializes a byte array back into an object using Java object
serialization.
+ *
+ * @param bytes the bytes to deserialize
+ * @return the deserialized object
+ * @throws RuntimeCamelException if deserialization fails
+ */
+ public static Object deserialize(byte[] bytes) {
+ try (ByteArrayInputStream bis = new ByteArrayInputStream(bytes);
+ ObjectInputStream ois = new ObjectInputStream(bis)) {
+ return ois.readObject();
+ } catch (IOException | ClassNotFoundException e) {
+ throw new RuntimeCamelException("Failed to deserialize value", e);
+ }
+ }
+
+ /**
+ * Deserializes an object from a portion of a byte array using Java object
serialization. Useful when the serialized
+ * data starts at an offset (e.g. after a protocol header).
+ *
+ * @param bytes the byte array containing the serialized
data
+ * @param offset the start offset within the array
+ * @param length the number of bytes to read
+ * @return the deserialized object
+ * @throws RuntimeCamelException if deserialization fails
+ */
+ public static Object deserialize(byte[] bytes, int offset, int length) {
+ try (ByteArrayInputStream bis = new ByteArrayInputStream(bytes,
offset, length);
+ ObjectInputStream ois = new ObjectInputStream(bis)) {
+ return ois.readObject();
+ } catch (IOException | ClassNotFoundException e) {
+ throw new RuntimeCamelException("Failed to deserialize value", e);
+ }
+ }
+
+ /**
+ * Deserializes an object from a {@link ByteBuffer} using Java object
serialization. The buffer's remaining bytes
+ * are consumed.
+ *
+ * @param buffer the ByteBuffer containing the serialized
bytes
+ * @return the deserialized object
+ * @throws RuntimeCamelException if deserialization fails
+ */
+ public static Object deserialize(ByteBuffer buffer) {
+ byte[] bytes = new byte[buffer.remaining()];
+ buffer.get(bytes);
+ return deserialize(bytes);
+ }
+}
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 c1853e4c172a..3b78af19a580 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", null);
+ kvRepository.put("aggregate:order-1", "exchange-holder");
// 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", null);
+ kvRepository.put("aggregate:order-1", "exchange-data");
// The idempotent entry should still resolve correctly
assertThat(idempotentRepository.contains("order-1")).isTrue();
