This is an automated email from the ASF dual-hosted git repository.
gnodet pushed a commit to branch
implement-the-keyvaluerepository-implementations-f
in repository https://gitbox.apache.org/repos/asf/camel.git
The following commit(s) were added to
refs/heads/implement-the-keyvaluerepository-implementations-f by this push:
new f385f666b3ba CAMEL-24463: Use shared KeyValueRepositoryHelper in all
backends
f385f666b3ba is described below
commit f385f666b3ba410ecfab23eebeb7cac542938968
Author: Guillaume Nodet <[email protected]>
AuthorDate: Tue Sep 1 16:22:38 2026 +0200
CAMEL-24463: Use shared KeyValueRepositoryHelper in all backends
Replace private serialize/deserialize methods in JDBC, JPA, Cassandra,
and Kafka KeyValueRepository implementations with shared
KeyValueRepositoryHelper from camel-support.
Co-Authored-By: Claude Opus 4.6 <[email protected]>
---
.../cassandra/CassandraKeyValueRepository.java | 51 +++-------------------
.../keyvalue/jpa/JpaKeyValueRepository.java | 41 ++++-------------
.../keyvalue/kafka/KafkaKeyValueRepository.java | 18 ++------
.../keyvalue/jdbc/JdbcKeyValueRepository.java | 47 ++------------------
4 files changed, 21 insertions(+), 136 deletions(-)
diff --git
a/components/camel-cassandraql/src/main/java/org/apache/camel/processor/keyvalue/cassandra/CassandraKeyValueRepository.java
b/components/camel-cassandraql/src/main/java/org/apache/camel/processor/keyvalue/cassandra/CassandraKeyValueRepository.java
index 3adfb53f3489..988aa164e86d 100644
---
a/components/camel-cassandraql/src/main/java/org/apache/camel/processor/keyvalue/cassandra/CassandraKeyValueRepository.java
+++
b/components/camel-cassandraql/src/main/java/org/apache/camel/processor/keyvalue/cassandra/CassandraKeyValueRepository.java
@@ -16,11 +16,6 @@
*/
package org.apache.camel.processor.keyvalue.cassandra;
-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 java.time.Duration;
import java.util.Collections;
@@ -36,13 +31,13 @@ import com.datastax.oss.driver.api.core.cql.SimpleStatement;
import com.datastax.oss.driver.api.querybuilder.delete.Delete;
import com.datastax.oss.driver.api.querybuilder.select.Select;
import com.datastax.oss.driver.api.querybuilder.truncate.Truncate;
-import org.apache.camel.RuntimeCamelException;
import org.apache.camel.api.management.ManagedAttribute;
import org.apache.camel.api.management.ManagedOperation;
import org.apache.camel.api.management.ManagedResource;
import org.apache.camel.spi.Configurer;
import org.apache.camel.spi.KeyValueRepository;
import org.apache.camel.spi.Metadata;
+import org.apache.camel.support.KeyValueRepositoryHelper;
import org.apache.camel.support.service.ServiceSupport;
import org.apache.camel.util.ObjectHelper;
import org.apache.camel.utils.cassandra.CassandraSessionHolder;
@@ -124,42 +119,6 @@ public class CassandraKeyValueRepository extends
ServiceSupport implements KeyVa
return row == null || row.getBoolean("[applied]");
}
- /**
- * Serializes an object to a {@link ByteBuffer} using Java serialization.
- *
- * @param value the object to serialize (must be {@link
java.io.Serializable})
- * @return a ByteBuffer containing the serialized
bytes
- * @throws RuntimeCamelException if serialization fails
- */
- private ByteBuffer serialize(Object value) {
- try (ByteArrayOutputStream baos = new ByteArrayOutputStream();
- ObjectOutputStream oos = new ObjectOutputStream(baos)) {
- oos.writeObject(value);
- oos.flush();
- return ByteBuffer.wrap(baos.toByteArray());
- } catch (IOException e) {
- throw new RuntimeCamelException("Failed to serialize value", e);
- }
- }
-
- /**
- * Deserializes an object from a {@link ByteBuffer} using Java
serialization.
- *
- * @param buffer the ByteBuffer containing the serialized
bytes
- * @return the deserialized object
- * @throws RuntimeCamelException if deserialization fails
- */
- private Object deserialize(ByteBuffer buffer) {
- byte[] bytes = new byte[buffer.remaining()];
- buffer.get(bytes);
- try (ByteArrayInputStream bais = new ByteArrayInputStream(bytes);
- ObjectInputStream ois = new ObjectInputStream(bais)) {
- return ois.readObject();
- } catch (IOException | ClassNotFoundException e) {
- throw new RuntimeCamelException("Failed to deserialize value", e);
- }
- }
-
//
-------------------------------------------------------------------------
// Lifecycle methods
@@ -276,7 +235,7 @@ public class CassandraKeyValueRepository extends
ServiceSupport implements KeyVa
return null;
}
ByteBuffer buffer = row.getByteBuffer(VALUE_COLUMN);
- return buffer != null ? deserialize(buffer) : null;
+ return buffer != null ? KeyValueRepositoryHelper.deserialize(buffer) :
null;
}
@Override
@@ -285,7 +244,7 @@ public class CassandraKeyValueRepository extends
ServiceSupport implements KeyVa
LOGGER.debug("Putting key {} with TTL {}", key, ttl);
// Read the previous value before upserting
Object oldValue = get(key);
- ByteBuffer serializedValue = serialize(value);
+ ByteBuffer serializedValue =
KeyValueRepositoryHelper.serializeToByteBuffer(value);
int ttlSeconds = toTtlSeconds(ttl);
if (ttlSeconds > 0) {
getSession().execute(insertWithTtlStatement.bind(key,
serializedValue, ttlSeconds));
@@ -343,7 +302,7 @@ public class CassandraKeyValueRepository extends
ServiceSupport implements KeyVa
@Override
public Object putIfAbsent(String key, Object value, Duration ttl) {
LOGGER.debug("Putting key {} if absent with TTL {}", key, ttl);
- ByteBuffer serializedValue = serialize(value);
+ ByteBuffer serializedValue =
KeyValueRepositoryHelper.serializeToByteBuffer(value);
ResultSet rs;
int ttlSeconds = toTtlSeconds(ttl);
if (ttlSeconds > 0) {
@@ -357,7 +316,7 @@ public class CassandraKeyValueRepository extends
ServiceSupport implements KeyVa
}
// Insert was not applied; return the existing value from the result
row
ByteBuffer existingBuffer = row.getByteBuffer(VALUE_COLUMN);
- return existingBuffer != null ? deserialize(existingBuffer) : null;
+ return existingBuffer != null ?
KeyValueRepositoryHelper.deserialize(existingBuffer) : null;
}
@Override
diff --git
a/components/camel-jpa/src/main/java/org/apache/camel/processor/keyvalue/jpa/JpaKeyValueRepository.java
b/components/camel-jpa/src/main/java/org/apache/camel/processor/keyvalue/jpa/JpaKeyValueRepository.java
index b9a0e9987fdb..849519394e47 100644
---
a/components/camel-jpa/src/main/java/org/apache/camel/processor/keyvalue/jpa/JpaKeyValueRepository.java
+++
b/components/camel-jpa/src/main/java/org/apache/camel/processor/keyvalue/jpa/JpaKeyValueRepository.java
@@ -16,11 +16,6 @@
*/
package org.apache.camel.processor.keyvalue.jpa;
-import java.io.ByteArrayInputStream;
-import java.io.ByteArrayOutputStream;
-import java.io.IOException;
-import java.io.ObjectInputStream;
-import java.io.ObjectOutputStream;
import java.time.Duration;
import java.util.Iterator;
import java.util.List;
@@ -31,7 +26,6 @@ import jakarta.persistence.EntityManagerFactory;
import jakarta.persistence.PersistenceException;
import jakarta.persistence.Query;
-import org.apache.camel.RuntimeCamelException;
import org.apache.camel.api.management.ManagedAttribute;
import org.apache.camel.api.management.ManagedOperation;
import org.apache.camel.api.management.ManagedResource;
@@ -40,6 +34,7 @@ import org.apache.camel.component.jpa.TransactionStrategy;
import org.apache.camel.spi.Configurer;
import org.apache.camel.spi.KeyValueRepository;
import org.apache.camel.spi.Metadata;
+import org.apache.camel.support.KeyValueRepositoryHelper;
import org.apache.camel.support.service.ServiceSupport;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -125,7 +120,7 @@ public class JpaKeyValueRepository extends ServiceSupport
implements KeyValueRep
entityManager.flush();
rc[0] = null;
} else {
- rc[0] = deserialize(entry.getItemValue());
+ rc[0] =
KeyValueRepositoryHelper.deserialize(entry.getItemValue());
}
} catch (Exception ex) {
String contextInfo = String.format(SOMETHING_WENT_WRONG,
ex.getMessage());
@@ -152,12 +147,12 @@ public class JpaKeyValueRepository extends ServiceSupport
implements KeyValueRep
}
try {
long expiresAt = toExpiresAt(ttl);
- byte[] serializedValue = serialize(value);
+ byte[] serializedValue =
KeyValueRepositoryHelper.serialize(value);
KeyValueEntry entry = findByKey(entityManager, key);
if (entry != null) {
if (!entry.isExpired()) {
- rc[0] = deserialize(entry.getItemValue());
+ rc[0] =
KeyValueRepositoryHelper.deserialize(entry.getItemValue());
}
entry.setItemValue(serializedValue);
entry.setExpiresAt(expiresAt);
@@ -199,7 +194,7 @@ public class JpaKeyValueRepository extends ServiceSupport
implements KeyValueRep
entityManager.flush();
rc[0] = null;
} else {
- rc[0] = deserialize(entry.getItemValue());
+ rc[0] =
KeyValueRepositoryHelper.deserialize(entry.getItemValue());
entityManager.remove(entry);
entityManager.flush();
}
@@ -322,12 +317,12 @@ public class JpaKeyValueRepository extends ServiceSupport
implements KeyValueRep
}
try {
long expiresAt = toExpiresAt(ttl);
- byte[] serializedValue = serialize(value);
+ byte[] serializedValue =
KeyValueRepositoryHelper.serialize(value);
KeyValueEntry entry = findByKey(entityManager, key);
if (entry != null && !entry.isExpired()) {
// key exists and is valid -- return existing value
- rc[0] = deserialize(entry.getItemValue());
+ rc[0] =
KeyValueRepositoryHelper.deserialize(entry.getItemValue());
} else if (entry != null) {
// key exists but expired -- update in place
entry.setItemValue(serializedValue);
@@ -349,7 +344,7 @@ public class JpaKeyValueRepository extends ServiceSupport
implements KeyValueRep
// re-read to return the existing value
try {
KeyValueEntry existing = findByKey(entityManager, key);
- rc[0] = existing != null ?
deserialize(existing.getItemValue()) : null;
+ rc[0] = existing != null ?
KeyValueRepositoryHelper.deserialize(existing.getItemValue()) : null;
} catch (Exception inner) {
// fall through with null
rc[0] = null;
@@ -490,26 +485,6 @@ public class JpaKeyValueRepository extends ServiceSupport
implements KeyValueRep
return System.currentTimeMillis() + ttl.toMillis();
}
- private 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);
- }
- }
-
- private static Object deserialize(byte[] data) {
- try (ByteArrayInputStream bis = new ByteArrayInputStream(data);
- ObjectInputStream ois = new ObjectInputStream(bis)) {
- return ois.readObject();
- } catch (IOException | ClassNotFoundException e) {
- throw new RuntimeCamelException("Failed to deserialize value", e);
- }
- }
-
private static void closeEntityManager(EntityManager entityManager) {
try {
if (entityManager.isOpen()) {
diff --git
a/components/camel-kafka/src/main/java/org/apache/camel/processor/keyvalue/kafka/KafkaKeyValueRepository.java
b/components/camel-kafka/src/main/java/org/apache/camel/processor/keyvalue/kafka/KafkaKeyValueRepository.java
index 4d15df912bc7..714ca455366c 100644
---
a/components/camel-kafka/src/main/java/org/apache/camel/processor/keyvalue/kafka/KafkaKeyValueRepository.java
+++
b/components/camel-kafka/src/main/java/org/apache/camel/processor/keyvalue/kafka/KafkaKeyValueRepository.java
@@ -16,11 +16,8 @@
*/
package org.apache.camel.processor.keyvalue.kafka;
-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 java.time.Duration;
import java.util.Collection;
@@ -45,6 +42,7 @@ import
org.apache.camel.processor.idempotent.kafka.KafkaConsumerUtil;
import org.apache.camel.spi.Configurer;
import org.apache.camel.spi.KeyValueRepository;
import org.apache.camel.spi.Metadata;
+import org.apache.camel.support.KeyValueRepositoryHelper;
import org.apache.camel.support.LRUCacheFactory;
import org.apache.camel.support.service.ServiceHelper;
import org.apache.camel.support.service.ServiceSupport;
@@ -306,9 +304,7 @@ public class KafkaKeyValueRepository extends ServiceSupport
implements KeyValueR
buf.putLong(expiresAt);
bos.write(buf.array());
// Write serialized value
- ObjectOutputStream oos = new ObjectOutputStream(bos);
- oos.writeObject(value);
- oos.flush();
+ bos.write(KeyValueRepositoryHelper.serialize(value));
return bos.toByteArray();
} catch (IOException e) {
throw new RuntimeCamelException("Failed to serialize value for
Kafka", e);
@@ -316,14 +312,8 @@ public class KafkaKeyValueRepository extends
ServiceSupport implements KeyValueR
}
private Object deserializeValue(byte[] data) {
- try {
- // Value starts at offset 9 (1 byte action + 8 bytes expiresAt)
- ByteArrayInputStream bis = new ByteArrayInputStream(data, 9,
data.length - 9);
- ObjectInputStream ois = new ObjectInputStream(bis);
- return ois.readObject();
- } catch (IOException | ClassNotFoundException e) {
- throw new RuntimeCamelException("Failed to deserialize value from
Kafka", e);
- }
+ // Value starts at offset 9 (1 byte action + 8 bytes expiresAt)
+ return KeyValueRepositoryHelper.deserialize(data, 9, data.length - 9);
}
private long deserializeExpiresAt(byte[] data) {
diff --git
a/components/camel-sql/src/main/java/org/apache/camel/processor/keyvalue/jdbc/JdbcKeyValueRepository.java
b/components/camel-sql/src/main/java/org/apache/camel/processor/keyvalue/jdbc/JdbcKeyValueRepository.java
index ff82631e3299..bcae88b515df 100644
---
a/components/camel-sql/src/main/java/org/apache/camel/processor/keyvalue/jdbc/JdbcKeyValueRepository.java
+++
b/components/camel-sql/src/main/java/org/apache/camel/processor/keyvalue/jdbc/JdbcKeyValueRepository.java
@@ -16,11 +16,6 @@
*/
package org.apache.camel.processor.keyvalue.jdbc;
-import java.io.ByteArrayInputStream;
-import java.io.ByteArrayOutputStream;
-import java.io.IOException;
-import java.io.ObjectInputStream;
-import java.io.ObjectOutputStream;
import java.time.Duration;
import java.util.Collections;
import java.util.LinkedHashSet;
@@ -29,13 +24,13 @@ import java.util.Set;
import javax.sql.DataSource;
-import org.apache.camel.RuntimeCamelException;
import org.apache.camel.api.management.ManagedAttribute;
import org.apache.camel.api.management.ManagedOperation;
import org.apache.camel.api.management.ManagedResource;
import org.apache.camel.spi.Configurer;
import org.apache.camel.spi.KeyValueRepository;
import org.apache.camel.spi.Metadata;
+import org.apache.camel.support.KeyValueRepositoryHelper;
import org.apache.camel.support.service.ServiceSupport;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -230,7 +225,7 @@ public class JdbcKeyValueRepository extends ServiceSupport
implements KeyValueRe
jdbcTemplate.update(getDeleteString(), key);
// insert the new row
long expiresAt = toExpiresAt(ttl);
- jdbcTemplate.update(getInsertString(), key, serialize(value),
expiresAt);
+ jdbcTemplate.update(getInsertString(), key,
KeyValueRepositoryHelper.serialize(value), expiresAt);
return oldValue;
});
}
@@ -281,7 +276,7 @@ public class JdbcKeyValueRepository extends ServiceSupport
implements KeyValueRe
// attempt to insert
long expiresAt = toExpiresAt(ttl);
try {
- jdbcTemplate.update(getInsertString(), key, serialize(value),
expiresAt);
+ jdbcTemplate.update(getInsertString(), key,
KeyValueRepositoryHelper.serialize(value), expiresAt);
return null;
} catch (DuplicateKeyException e) {
// concurrent insert race -- another thread/node won
@@ -317,7 +312,7 @@ public class JdbcKeyValueRepository extends ServiceSupport
implements KeyValueRe
jdbcTemplate.update(getDeleteString(), key);
return null;
}
- return deserialize(bytes);
+ return KeyValueRepositoryHelper.deserialize(bytes);
}, key);
} catch (EmptyResultDataAccessException e) {
return null;
@@ -331,40 +326,6 @@ public class JdbcKeyValueRepository extends ServiceSupport
implements KeyValueRe
return System.currentTimeMillis() + ttl.toMillis();
}
- /**
- * Serializes an object to a byte array using Java object serialization.
- *
- * @param value the object to serialize
- * @return the serialized bytes
- * @throws RuntimeCamelException if serialization fails
- */
- private 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);
- }
- }
-
- /**
- * 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
- */
- private 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);
- }
- }
-
// ---- Getters and Setters ----
@ManagedAttribute(description = "The name of the database table")
