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
commit 11918a6244280788c6a3b790fd957ed5ca1b0c55 Author: Guillaume Nodet <[email protected]> AuthorDate: Tue Sep 1 10:26:05 2026 +0200 CAMEL-24463: Add KeyValueRepository implementations for JDBC, JPA, Cassandra, and Kafka Add concrete KeyValueRepository implementations for four persistence backends: - JdbcKeyValueRepository (camel-sql): Uses Spring JdbcTemplate/TransactionTemplate with auto-creating table, lazy TTL eviction via EXPIRES_AT column - JpaKeyValueRepository (camel-jpa): Uses JPA EntityManager with KeyValueEntry entity, JPQL queries, and lazy TTL eviction - CassandraKeyValueRepository (camel-cassandraql): Uses native Cassandra TTL via USING TTL clause, lightweight transactions for putIfAbsent - KafkaKeyValueRepository (camel-kafka): Uses local LRU cache with Kafka topic as changelog for distributed state synchronization Includes unit tests for JDBC implementation (22 tests passing). Co-Authored-By: Claude Opus 4.6 <[email protected]> --- .../CassandraKeyValueRepositoryConfigurer.java | 63 ++ .../services/org/apache/camel/bean.properties | 2 +- .../camel/bean/CassandraKeyValueRepository.json | 16 + ....keyvalue.cassandra.CassandraKeyValueRepository | 2 + .../cassandra/CassandraKeyValueRepository.java | 402 +++++++++++++ .../jpa/JpaKeyValueRepositoryConfigurer.java | 69 +++ .../services/org/apache/camel/bean.properties | 7 + .../apache/camel/bean/JpaKeyValueRepository.json | 16 + ...el.processor.keyvalue.jpa.JpaKeyValueRepository | 2 + .../keyvalue/jpa/JpaKeyValueRepository.java | 528 +++++++++++++++++ .../processor/keyvalue/jpa/KeyValueEntry.java | 108 ++++ .../kafka/KafkaKeyValueRepositoryConfigurer.java | 90 +++ .../services/org/apache/camel/bean.properties | 2 +- .../apache/camel/bean/KafkaKeyValueRepository.json | 16 + ...rocessor.keyvalue.kafka.KafkaKeyValueRepository | 2 + .../keyvalue/kafka/KafkaKeyValueRepository.java | 645 +++++++++++++++++++++ .../jdbc/JdbcKeyValueRepositoryConfigurer.java | 117 ++++ .../services/org/apache/camel/bean.properties | 2 +- .../apache/camel/bean/JdbcKeyValueRepository.json | 16 + ....processor.keyvalue.jdbc.JdbcKeyValueRepository | 2 + .../keyvalue/jdbc/JdbcKeyValueRepository.java | 470 +++++++++++++++ .../keyvalue/jdbc/JdbcKeyValueRepositoryTest.java | 243 ++++++++ 22 files changed, 2817 insertions(+), 3 deletions(-) diff --git a/components/camel-cassandraql/src/generated/java/org/apache/camel/processor/keyvalue/cassandra/CassandraKeyValueRepositoryConfigurer.java b/components/camel-cassandraql/src/generated/java/org/apache/camel/processor/keyvalue/cassandra/CassandraKeyValueRepositoryConfigurer.java new file mode 100644 index 000000000000..88dceac9d741 --- /dev/null +++ b/components/camel-cassandraql/src/generated/java/org/apache/camel/processor/keyvalue/cassandra/CassandraKeyValueRepositoryConfigurer.java @@ -0,0 +1,63 @@ +/* Generated by camel build tools - do NOT edit this file! */ +package org.apache.camel.processor.keyvalue.cassandra; + +import javax.annotation.processing.Generated; +import java.util.Map; + +import org.apache.camel.CamelContext; +import org.apache.camel.spi.ExtendedPropertyConfigurerGetter; +import org.apache.camel.spi.PropertyConfigurerGetter; +import org.apache.camel.spi.ConfigurerStrategy; +import org.apache.camel.spi.GeneratedPropertyConfigurer; +import org.apache.camel.util.CaseInsensitiveMap; +import org.apache.camel.processor.keyvalue.cassandra.CassandraKeyValueRepository; + +/** + * Generated by camel build tools - do NOT edit this file! + */ +@Generated("org.apache.camel.maven.packaging.GenerateConfigurerMojo") +@SuppressWarnings("unchecked") +public class CassandraKeyValueRepositoryConfigurer extends org.apache.camel.support.component.PropertyConfigurerSupport implements GeneratedPropertyConfigurer, PropertyConfigurerGetter { + + @Override + public boolean configure(CamelContext camelContext, Object obj, String name, Object value, boolean ignoreCase) { + org.apache.camel.processor.keyvalue.cassandra.CassandraKeyValueRepository target = (org.apache.camel.processor.keyvalue.cassandra.CassandraKeyValueRepository) obj; + switch (ignoreCase ? name.toLowerCase() : name) { + case "readconsistencylevel": + case "readConsistencyLevel": target.setReadConsistencyLevel(property(camelContext, com.datastax.oss.driver.api.core.ConsistencyLevel.class, value)); return true; + case "session": target.setSession(property(camelContext, com.datastax.oss.driver.api.core.CqlSession.class, value)); return true; + case "table": target.setTable(property(camelContext, java.lang.String.class, value)); return true; + case "writeconsistencylevel": + case "writeConsistencyLevel": target.setWriteConsistencyLevel(property(camelContext, com.datastax.oss.driver.api.core.ConsistencyLevel.class, value)); return true; + default: return false; + } + } + + @Override + public Class<?> getOptionType(String name, boolean ignoreCase) { + switch (ignoreCase ? name.toLowerCase() : name) { + case "readconsistencylevel": + case "readConsistencyLevel": return com.datastax.oss.driver.api.core.ConsistencyLevel.class; + case "session": return com.datastax.oss.driver.api.core.CqlSession.class; + case "table": return java.lang.String.class; + case "writeconsistencylevel": + case "writeConsistencyLevel": return com.datastax.oss.driver.api.core.ConsistencyLevel.class; + default: return null; + } + } + + @Override + public Object getOptionValue(Object obj, String name, boolean ignoreCase) { + org.apache.camel.processor.keyvalue.cassandra.CassandraKeyValueRepository target = (org.apache.camel.processor.keyvalue.cassandra.CassandraKeyValueRepository) obj; + switch (ignoreCase ? name.toLowerCase() : name) { + case "readconsistencylevel": + case "readConsistencyLevel": return target.getReadConsistencyLevel(); + case "session": return target.getSession(); + case "table": return target.getTable(); + case "writeconsistencylevel": + case "writeConsistencyLevel": return target.getWriteConsistencyLevel(); + default: return null; + } + } +} + diff --git a/components/camel-cassandraql/src/generated/resources/META-INF/services/org/apache/camel/bean.properties b/components/camel-cassandraql/src/generated/resources/META-INF/services/org/apache/camel/bean.properties index 4a3c9a72d760..302a1d7bf32f 100644 --- a/components/camel-cassandraql/src/generated/resources/META-INF/services/org/apache/camel/bean.properties +++ b/components/camel-cassandraql/src/generated/resources/META-INF/services/org/apache/camel/bean.properties @@ -1,5 +1,5 @@ # Generated by camel build tools - do NOT edit this file! -bean=CassandraAggregationRepository CassandraIdempotentRepository +bean=CassandraAggregationRepository CassandraIdempotentRepository CassandraKeyValueRepository groupId=org.apache.camel artifactId=camel-cassandraql version=4.23.0-SNAPSHOT diff --git a/components/camel-cassandraql/src/generated/resources/META-INF/services/org/apache/camel/bean/CassandraKeyValueRepository.json b/components/camel-cassandraql/src/generated/resources/META-INF/services/org/apache/camel/bean/CassandraKeyValueRepository.json new file mode 100644 index 000000000000..721fb2f727d1 --- /dev/null +++ b/components/camel-cassandraql/src/generated/resources/META-INF/services/org/apache/camel/bean/CassandraKeyValueRepository.json @@ -0,0 +1,16 @@ +{ + "bean": { + "kind": "bean", + "name": "CassandraKeyValueRepository", + "javaType": "org.apache.camel.processor.keyvalue.cassandra.CassandraKeyValueRepository", + "interfaceType": "org.apache.camel.spi.KeyValueRepository", + "title": "Cassandra Key Value Repository", + "description": "A Cassandra-based KeyValueRepository that uses a Cassandra table to store key-value entries. Advice: use LeveledCompaction for this table and tune read\/write consistency levels.", + "deprecated": false, + "groupId": "org.apache.camel", + "artifactId": "camel-cassandraql", + "version": "4.23.0-SNAPSHOT", + "properties": { "session": { "index": 0, "kind": "property", "displayName": "Session", "required": true, "type": "object", "javaType": "org.apache.camel.utils.cassandra.CassandraSessionHolder", "deprecated": false, "autowired": false, "secret": false, "description": "Cassandra session" }, "table": { "index": 1, "kind": "property", "displayName": "Table", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "autowired": false, "secret": false, "def [...] + } +} + diff --git a/components/camel-cassandraql/src/generated/resources/META-INF/services/org/apache/camel/configurer/org.apache.camel.processor.keyvalue.cassandra.CassandraKeyValueRepository b/components/camel-cassandraql/src/generated/resources/META-INF/services/org/apache/camel/configurer/org.apache.camel.processor.keyvalue.cassandra.CassandraKeyValueRepository new file mode 100644 index 000000000000..3f6ed2c0566b --- /dev/null +++ b/components/camel-cassandraql/src/generated/resources/META-INF/services/org/apache/camel/configurer/org.apache.camel.processor.keyvalue.cassandra.CassandraKeyValueRepository @@ -0,0 +1,2 @@ +# Generated by camel build tools - do NOT edit this file! +class=org.apache.camel.processor.keyvalue.cassandra.CassandraKeyValueRepositoryConfigurer 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 new file mode 100644 index 000000000000..80ab31b8590f --- /dev/null +++ b/components/camel-cassandraql/src/main/java/org/apache/camel/processor/keyvalue/cassandra/CassandraKeyValueRepository.java @@ -0,0 +1,402 @@ +/* + * 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.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.util.Collections; +import java.util.LinkedHashSet; +import java.util.Set; + +import com.datastax.oss.driver.api.core.ConsistencyLevel; +import com.datastax.oss.driver.api.core.CqlSession; +import com.datastax.oss.driver.api.core.cql.PreparedStatement; +import com.datastax.oss.driver.api.core.cql.ResultSet; +import com.datastax.oss.driver.api.core.cql.Row; +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.service.ServiceSupport; +import org.apache.camel.util.ObjectHelper; +import org.apache.camel.utils.cassandra.CassandraSessionHolder; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import static com.datastax.oss.driver.api.querybuilder.QueryBuilder.bindMarker; +import static com.datastax.oss.driver.api.querybuilder.QueryBuilder.insertInto; +import static org.apache.camel.utils.cassandra.CassandraUtils.applyConsistencyLevel; +import static org.apache.camel.utils.cassandra.CassandraUtils.generateDelete; +import static org.apache.camel.utils.cassandra.CassandraUtils.generateSelect; +import static org.apache.camel.utils.cassandra.CassandraUtils.generateTruncate; + +/** + * A Cassandra-based implementation of {@link KeyValueRepository} that stores key-value entries in a Cassandra table. + * <p/> + * Values are serialized to bytes using Java {@link ObjectOutputStream} and stored in a {@code BLOB} column. Keys are + * stored as {@code TEXT}. Time-to-live is handled natively by Cassandra's {@code USING TTL} clause on {@code INSERT} + * statements, so expired entries are removed automatically by Cassandra without any client-side eviction logic. + * <p/> + * The {@link #putIfAbsent(String, Object, long)} method is implemented atomically using Cassandra's lightweight + * transactions ({@code INSERT ... IF NOT EXISTS}). + * <p/> + * Advice: use LeveledCompaction for the backing table and tune read/write consistency levels for your use case. + * + * @since 4.23 + */ +@Metadata(label = "bean", + description = "A Cassandra-based KeyValueRepository that uses a Cassandra table to store key-value entries." + + " Advice: use LeveledCompaction for this table and tune read/write consistency levels.", + annotations = { "interfaceName=org.apache.camel.spi.KeyValueRepository" }) +@Configurer(metadataOnly = true) +@ManagedResource(description = "Cassandra based key-value repository") +public class CassandraKeyValueRepository extends ServiceSupport implements KeyValueRepository { + + private static final Logger LOGGER = LoggerFactory.getLogger(CassandraKeyValueRepository.class); + + private static final String KEY_COLUMN = "KEY"; + private static final String VALUE_COLUMN = "VALUE"; + + @Metadata(description = "Cassandra session", required = true) + private CassandraSessionHolder session; + @Metadata(description = "The table name for storing the data", defaultValue = "CAMEL_KEYVALUE") + private String table = "CAMEL_KEYVALUE"; + @Metadata(description = "Write consistency level", + enums = "ANY,ONE,TWO,THREE,QUORUM,ALL,LOCAL_ONE,LOCAL_QUORUM,EACH_QUORUM,SERIAL,LOCAL_SERIAL") + private ConsistencyLevel writeConsistencyLevel; + @Metadata(description = "Read consistency level", + enums = "ANY,ONE,TWO,THREE,QUORUM,ALL,LOCAL_ONE,LOCAL_QUORUM,EACH_QUORUM,SERIAL,LOCAL_SERIAL") + private ConsistencyLevel readConsistencyLevel; + + private PreparedStatement insertStatement; + private PreparedStatement insertWithTtlStatement; + private PreparedStatement selectStatement; + private PreparedStatement deleteStatement; + private PreparedStatement selectAllKeysStatement; + private PreparedStatement truncateStatement; + private PreparedStatement insertIfNotExistsStatement; + private PreparedStatement insertIfNotExistsWithTtlStatement; + + public CassandraKeyValueRepository() { + } + + public CassandraKeyValueRepository(CqlSession session) { + this.session = new CassandraSessionHolder(session); + } + + // ------------------------------------------------------------------------- + // Helper methods + + /** + * Checks whether a lightweight transaction was applied. + * + * @param resultSet the result set from a conditional statement + * @return {@code true} if the statement was applied or the result is empty + */ + protected final boolean isApplied(ResultSet resultSet) { + Row row = resultSet.one(); + 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 + + @Override + protected void doStart() throws Exception { + ObjectHelper.notNull(session, "session", this); + session.start(); + initInsertStatement(); + initInsertWithTtlStatement(); + initSelectStatement(); + initDeleteStatement(); + initSelectAllKeysStatement(); + initClearStatement(); + initInsertIfNotExistsStatement(); + initInsertIfNotExistsWithTtlStatement(); + } + + @Override + protected void doStop() throws Exception { + if (session != null) { + session.stop(); + } + } + + // ------------------------------------------------------------------------- + // Prepared statement initialization + + protected void initInsertStatement() { + SimpleStatement statement = applyConsistencyLevel( + insertInto(table) + .value(KEY_COLUMN, bindMarker()) + .value(VALUE_COLUMN, bindMarker()) + .build(), + writeConsistencyLevel); + LOGGER.debug("Generated Insert {}", statement); + insertStatement = getSession().prepare(statement); + } + + protected void initInsertWithTtlStatement() { + SimpleStatement statement = applyConsistencyLevel( + insertInto(table) + .value(KEY_COLUMN, bindMarker()) + .value(VALUE_COLUMN, bindMarker()) + .usingTtl(bindMarker()) + .build(), + writeConsistencyLevel); + LOGGER.debug("Generated Insert with TTL {}", statement); + insertWithTtlStatement = getSession().prepare(statement); + } + + protected void initSelectStatement() { + Select select = generateSelect(table, new String[] { VALUE_COLUMN }, new String[] { KEY_COLUMN }); + SimpleStatement statement = applyConsistencyLevel(select.build(), readConsistencyLevel); + LOGGER.debug("Generated Select {}", statement); + selectStatement = getSession().prepare(statement); + } + + protected void initDeleteStatement() { + Delete delete = generateDelete(table, new String[] { KEY_COLUMN }, true); + SimpleStatement statement = applyConsistencyLevel(delete.build(), writeConsistencyLevel); + LOGGER.debug("Generated Delete {}", statement); + deleteStatement = getSession().prepare(statement); + } + + protected void initSelectAllKeysStatement() { + Select select = generateSelect(table, new String[] { KEY_COLUMN }, null); + SimpleStatement statement = applyConsistencyLevel(select.build(), readConsistencyLevel); + LOGGER.debug("Generated Select all keys {}", statement); + selectAllKeysStatement = getSession().prepare(statement); + } + + protected void initClearStatement() { + Truncate truncate = generateTruncate(table); + SimpleStatement statement = applyConsistencyLevel(truncate.build(), writeConsistencyLevel); + LOGGER.debug("Generated truncate for clear operation {}", statement); + truncateStatement = getSession().prepare(statement); + } + + protected void initInsertIfNotExistsStatement() { + SimpleStatement statement = applyConsistencyLevel( + insertInto(table) + .value(KEY_COLUMN, bindMarker()) + .value(VALUE_COLUMN, bindMarker()) + .ifNotExists() + .build(), + writeConsistencyLevel); + LOGGER.debug("Generated Insert if not exists {}", statement); + insertIfNotExistsStatement = getSession().prepare(statement); + } + + protected void initInsertIfNotExistsWithTtlStatement() { + SimpleStatement statement = applyConsistencyLevel( + insertInto(table) + .value(KEY_COLUMN, bindMarker()) + .value(VALUE_COLUMN, bindMarker()) + .ifNotExists() + .usingTtl(bindMarker()) + .build(), + writeConsistencyLevel); + LOGGER.debug("Generated Insert if not exists with TTL {}", statement); + insertIfNotExistsWithTtlStatement = getSession().prepare(statement); + } + + // ------------------------------------------------------------------------- + // KeyValueRepository operations + + @Override + @ManagedOperation(description = "Get value by key") + public Object get(String key) { + LOGGER.debug("Getting key {}", key); + ResultSet rs = getSession().execute(selectStatement.bind(key)); + Row row = rs.one(); + if (row == null) { + return null; + } + ByteBuffer buffer = row.getByteBuffer(VALUE_COLUMN); + return buffer != null ? deserialize(buffer) : null; + } + + @Override + @ManagedOperation(description = "Put a key-value pair with optional TTL") + public Object put(String key, Object value, long ttlMillis) { + LOGGER.debug("Putting key {} with TTL {} ms", key, ttlMillis); + // Read the previous value before upserting + Object oldValue = get(key); + ByteBuffer serializedValue = serialize(value); + int ttlSeconds = (int) (ttlMillis / 1000); + if (ttlSeconds > 0) { + getSession().execute(insertWithTtlStatement.bind(key, serializedValue, ttlSeconds)); + } else { + getSession().execute(insertStatement.bind(key, serializedValue)); + } + return oldValue; + } + + @Override + @ManagedOperation(description = "Delete a key") + public Object delete(String key) { + LOGGER.debug("Deleting key {}", key); + // Read the previous value before deleting + Object oldValue = get(key); + getSession().execute(deleteStatement.bind(key)); + return oldValue; + } + + @Override + @ManagedOperation(description = "Check if key exists") + public boolean contains(String key) { + LOGGER.debug("Checking key {}", key); + ResultSet rs = getSession().execute(selectStatement.bind(key)); + return rs.one() != null; + } + + @Override + public Set<String> keys() { + LOGGER.debug("Getting all keys from table {}", table); + ResultSet rs = getSession().execute(selectAllKeysStatement.bind()); + Set<String> result = new LinkedHashSet<>(); + for (Row row : rs) { + result.add(row.getString(KEY_COLUMN)); + } + return Collections.unmodifiableSet(result); + } + + @Override + @ManagedOperation(description = "Clear all entries") + public void clear() { + LOGGER.debug("Clear table {}", table); + getSession().execute(truncateStatement.bind()); + } + + /** + * Atomically stores the value under the given key only if no mapping already exists, using Cassandra's lightweight + * transaction ({@code INSERT ... IF NOT EXISTS}). + * + * @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 + */ + @Override + public Object putIfAbsent(String key, Object value, long ttlMillis) { + LOGGER.debug("Putting key {} if absent with TTL {} ms", key, ttlMillis); + ByteBuffer serializedValue = serialize(value); + ResultSet rs; + int ttlSeconds = (int) (ttlMillis / 1000); + if (ttlSeconds > 0) { + rs = getSession().execute(insertIfNotExistsWithTtlStatement.bind(key, serializedValue, ttlSeconds)); + } else { + rs = getSession().execute(insertIfNotExistsStatement.bind(key, serializedValue)); + } + Row row = rs.one(); + if (row == null || row.getBoolean("[applied]")) { + return null; + } + // Insert was not applied; return the existing value from the result row + ByteBuffer existingBuffer = row.getByteBuffer(VALUE_COLUMN); + return existingBuffer != null ? deserialize(existingBuffer) : null; + } + + @Override + @ManagedAttribute(description = "The number of entries in the repository") + public int size() { + return keys().size(); + } + + // ------------------------------------------------------------------------- + // Getters & Setters + + public CqlSession getSession() { + return session.getSession(); + } + + public void setSession(CqlSession session) { + this.session = new CassandraSessionHolder(session); + } + + public String getTable() { + return table; + } + + public void setTable(String table) { + this.table = table; + } + + public ConsistencyLevel getWriteConsistencyLevel() { + return writeConsistencyLevel; + } + + public void setWriteConsistencyLevel(ConsistencyLevel writeConsistencyLevel) { + this.writeConsistencyLevel = writeConsistencyLevel; + } + + public ConsistencyLevel getReadConsistencyLevel() { + return readConsistencyLevel; + } + + public void setReadConsistencyLevel(ConsistencyLevel readConsistencyLevel) { + this.readConsistencyLevel = readConsistencyLevel; + } +} diff --git a/components/camel-jpa/src/generated/java/org/apache/camel/processor/keyvalue/jpa/JpaKeyValueRepositoryConfigurer.java b/components/camel-jpa/src/generated/java/org/apache/camel/processor/keyvalue/jpa/JpaKeyValueRepositoryConfigurer.java new file mode 100644 index 000000000000..c7a43f4cd484 --- /dev/null +++ b/components/camel-jpa/src/generated/java/org/apache/camel/processor/keyvalue/jpa/JpaKeyValueRepositoryConfigurer.java @@ -0,0 +1,69 @@ +/* Generated by camel build tools - do NOT edit this file! */ +package org.apache.camel.processor.keyvalue.jpa; + +import javax.annotation.processing.Generated; +import java.util.Map; + +import org.apache.camel.CamelContext; +import org.apache.camel.spi.ExtendedPropertyConfigurerGetter; +import org.apache.camel.spi.PropertyConfigurerGetter; +import org.apache.camel.spi.ConfigurerStrategy; +import org.apache.camel.spi.GeneratedPropertyConfigurer; +import org.apache.camel.util.CaseInsensitiveMap; +import org.apache.camel.processor.keyvalue.jpa.JpaKeyValueRepository; + +/** + * Generated by camel build tools - do NOT edit this file! + */ +@Generated("org.apache.camel.maven.packaging.GenerateConfigurerMojo") +@SuppressWarnings("unchecked") +public class JpaKeyValueRepositoryConfigurer extends org.apache.camel.support.component.PropertyConfigurerSupport implements GeneratedPropertyConfigurer, PropertyConfigurerGetter { + + @Override + public boolean configure(CamelContext camelContext, Object obj, String name, Object value, boolean ignoreCase) { + org.apache.camel.processor.keyvalue.jpa.JpaKeyValueRepository target = (org.apache.camel.processor.keyvalue.jpa.JpaKeyValueRepository) obj; + switch (ignoreCase ? name.toLowerCase() : name) { + case "entitymanagerfactory": + case "entityManagerFactory": target.setEntityManagerFactory(property(camelContext, jakarta.persistence.EntityManagerFactory.class, value)); return true; + case "jointransaction": + case "joinTransaction": target.setJoinTransaction(property(camelContext, boolean.class, value)); return true; + case "sharedentitymanager": + case "sharedEntityManager": target.setSharedEntityManager(property(camelContext, boolean.class, value)); return true; + case "transactionstrategy": + case "transactionStrategy": target.setTransactionStrategy(property(camelContext, org.apache.camel.component.jpa.TransactionStrategy.class, value)); return true; + default: return false; + } + } + + @Override + public Class<?> getOptionType(String name, boolean ignoreCase) { + switch (ignoreCase ? name.toLowerCase() : name) { + case "entitymanagerfactory": + case "entityManagerFactory": return jakarta.persistence.EntityManagerFactory.class; + case "jointransaction": + case "joinTransaction": return boolean.class; + case "sharedentitymanager": + case "sharedEntityManager": return boolean.class; + case "transactionstrategy": + case "transactionStrategy": return org.apache.camel.component.jpa.TransactionStrategy.class; + default: return null; + } + } + + @Override + public Object getOptionValue(Object obj, String name, boolean ignoreCase) { + org.apache.camel.processor.keyvalue.jpa.JpaKeyValueRepository target = (org.apache.camel.processor.keyvalue.jpa.JpaKeyValueRepository) obj; + switch (ignoreCase ? name.toLowerCase() : name) { + case "entitymanagerfactory": + case "entityManagerFactory": return target.getEntityManagerFactory(); + case "jointransaction": + case "joinTransaction": return target.isJoinTransaction(); + case "sharedentitymanager": + case "sharedEntityManager": return target.isSharedEntityManager(); + case "transactionstrategy": + case "transactionStrategy": return target.getTransactionStrategy(); + default: return null; + } + } +} + diff --git a/components/camel-jpa/src/generated/resources/META-INF/services/org/apache/camel/bean.properties b/components/camel-jpa/src/generated/resources/META-INF/services/org/apache/camel/bean.properties new file mode 100644 index 000000000000..188d375e61dc --- /dev/null +++ b/components/camel-jpa/src/generated/resources/META-INF/services/org/apache/camel/bean.properties @@ -0,0 +1,7 @@ +# Generated by camel build tools - do NOT edit this file! +bean=JpaKeyValueRepository +groupId=org.apache.camel +artifactId=camel-jpa +version=4.23.0-SNAPSHOT +projectName=Camel :: JPA +projectDescription=Camel JPA support diff --git a/components/camel-jpa/src/generated/resources/META-INF/services/org/apache/camel/bean/JpaKeyValueRepository.json b/components/camel-jpa/src/generated/resources/META-INF/services/org/apache/camel/bean/JpaKeyValueRepository.json new file mode 100644 index 000000000000..2ec1140fe36b --- /dev/null +++ b/components/camel-jpa/src/generated/resources/META-INF/services/org/apache/camel/bean/JpaKeyValueRepository.json @@ -0,0 +1,16 @@ +{ + "bean": { + "kind": "bean", + "name": "JpaKeyValueRepository", + "javaType": "org.apache.camel.processor.keyvalue.jpa.JpaKeyValueRepository", + "interfaceType": "org.apache.camel.spi.KeyValueRepository", + "title": "Jpa Key Value Repository", + "description": "A JPA-based KeyValueRepository that stores entries using JPA entities.", + "deprecated": false, + "groupId": "org.apache.camel", + "artifactId": "camel-jpa", + "version": "4.23.0-SNAPSHOT", + "properties": { "entityManagerFactory": { "index": 0, "kind": "property", "displayName": "Entity Manager Factory", "required": true, "type": "object", "javaType": "jakarta.persistence.EntityManagerFactory", "deprecated": false, "autowired": false, "secret": false, "description": "The JPA EntityManagerFactory to use" }, "transactionStrategy": { "index": 1, "kind": "property", "displayName": "Transaction Strategy", "required": false, "type": "object", "javaType": "org.apache.camel.comp [...] + } +} + diff --git a/components/camel-jpa/src/generated/resources/META-INF/services/org/apache/camel/configurer/org.apache.camel.processor.keyvalue.jpa.JpaKeyValueRepository b/components/camel-jpa/src/generated/resources/META-INF/services/org/apache/camel/configurer/org.apache.camel.processor.keyvalue.jpa.JpaKeyValueRepository new file mode 100644 index 000000000000..605b383a08e7 --- /dev/null +++ b/components/camel-jpa/src/generated/resources/META-INF/services/org/apache/camel/configurer/org.apache.camel.processor.keyvalue.jpa.JpaKeyValueRepository @@ -0,0 +1,2 @@ +# Generated by camel build tools - do NOT edit this file! +class=org.apache.camel.processor.keyvalue.jpa.JpaKeyValueRepositoryConfigurer 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 new file mode 100644 index 000000000000..aaa80c9fcc83 --- /dev/null +++ b/components/camel-jpa/src/main/java/org/apache/camel/processor/keyvalue/jpa/JpaKeyValueRepository.java @@ -0,0 +1,528 @@ +/* + * 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.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.util.Iterator; +import java.util.List; +import java.util.Set; + +import jakarta.persistence.EntityManager; +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; +import org.apache.camel.component.jpa.DefaultTransactionStrategy; +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.service.ServiceSupport; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import static org.apache.camel.component.jpa.JpaHelper.getTargetEntityManager; + +/** + * A JPA-based {@link KeyValueRepository} that stores entries as {@link KeyValueEntry} entities. + * <p/> + * Values are serialized to byte arrays using Java Object Serialization and stored in a {@code CAMEL_KEYVALUE} table. + * Expired entries are cleaned up lazily on access. + * <p/> + * This implementation follows the same transaction and entity-manager patterns used by + * {@link org.apache.camel.processor.idempotent.jpa.JpaMessageIdRepository}. + * + * @since 4.23 + */ +@Metadata(label = "bean", + description = "A JPA-based KeyValueRepository that stores entries using JPA entities.", + annotations = { "interfaceName=org.apache.camel.spi.KeyValueRepository" }) +@Configurer(metadataOnly = true) +@ManagedResource(description = "JPA based key-value repository") +public class JpaKeyValueRepository extends ServiceSupport implements KeyValueRepository { + + private static final String QUERY_BY_KEY + = "select e from " + KeyValueEntry.class.getName() + " e where e.itemKey = ?1"; + private static final String QUERY_ALL_KEYS + = "select e.itemKey from " + KeyValueEntry.class.getName() + " e where e.expiresAt = 0 or e.expiresAt > ?1"; + private static final String QUERY_ALL + = "select e from " + KeyValueEntry.class.getName() + " e"; + private static final String QUERY_COUNT_VALID + = "select count(e) from " + KeyValueEntry.class.getName() + " e where e.expiresAt = 0 or e.expiresAt > ?1"; + + private static final String SOMETHING_WENT_WRONG + = "Something went wrong in JpaKeyValueRepository: %s"; + + private static final Logger LOG = LoggerFactory.getLogger(JpaKeyValueRepository.class); + + @Metadata(description = "The JPA EntityManagerFactory to use", required = true) + private EntityManagerFactory entityManagerFactory; + @Metadata(description = "The TransactionStrategy to use for transactional operations") + private TransactionStrategy transactionStrategy; + @Metadata(description = "Whether to join an existing transaction", defaultValue = "true") + private boolean joinTransaction = true; + @Metadata(description = "Whether to use a shared EntityManager", defaultValue = "false") + private boolean sharedEntityManager; + + /** + * Creates a new JPA key-value repository. The {@link #setEntityManagerFactory(EntityManagerFactory)} must be called + * before starting. + */ + public JpaKeyValueRepository() { + } + + /** + * Creates a new JPA key-value repository with the given entity manager factory. + * + * @param entityManagerFactory the JPA entity manager factory to use + */ + public JpaKeyValueRepository(EntityManagerFactory entityManagerFactory) { + this.entityManagerFactory = entityManagerFactory; + } + + // ---- KeyValueRepository implementation ---- + + @Override + @ManagedOperation(description = "Get value by key") + public Object get(String key) { + final Object[] rc = new Object[1]; + final EntityManager entityManager + = getTargetEntityManager(null, entityManagerFactory, false, sharedEntityManager, true); + + transactionStrategy.executeInTransaction(() -> { + if (joinTransaction) { + entityManager.joinTransaction(); + } + try { + KeyValueEntry entry = findByKey(entityManager, key); + if (entry == null) { + rc[0] = null; + } else if (entry.isExpired()) { + entityManager.remove(entry); + entityManager.flush(); + rc[0] = null; + } else { + rc[0] = deserialize(entry.getItemValue()); + } + } catch (Exception ex) { + String contextInfo = String.format(SOMETHING_WENT_WRONG, ex.getMessage()); + throw new PersistenceException(contextInfo, ex); + } finally { + closeEntityManager(entityManager); + } + }); + + LOG.debug("get {} -> {}", key, rc[0] != null ? "found" : "null"); + return rc[0]; + } + + @Override + @ManagedOperation(description = "Put a key-value pair with optional TTL") + public Object put(String key, Object value, long ttlMillis) { + final Object[] rc = new Object[1]; + final EntityManager entityManager + = getTargetEntityManager(null, entityManagerFactory, false, sharedEntityManager, true); + + transactionStrategy.executeInTransaction(() -> { + if (joinTransaction) { + entityManager.joinTransaction(); + } + try { + long expiresAt = ttlMillis > 0 ? System.currentTimeMillis() + ttlMillis : 0; + byte[] serializedValue = serialize(value); + + KeyValueEntry entry = findByKey(entityManager, key); + if (entry != null) { + if (!entry.isExpired()) { + rc[0] = deserialize(entry.getItemValue()); + } + entry.setItemValue(serializedValue); + entry.setExpiresAt(expiresAt); + entityManager.merge(entry); + } else { + entry = new KeyValueEntry(key, serializedValue, expiresAt); + entityManager.persist(entry); + } + entityManager.flush(); + } catch (Exception ex) { + String contextInfo = String.format(SOMETHING_WENT_WRONG, ex.getMessage()); + throw new PersistenceException(contextInfo, ex); + } finally { + closeEntityManager(entityManager); + } + }); + + LOG.debug("put {} -> previous={}", key, rc[0] != null ? "found" : "null"); + return rc[0]; + } + + @Override + @ManagedOperation(description = "Delete a key") + public Object delete(String key) { + final Object[] rc = new Object[1]; + final EntityManager entityManager + = getTargetEntityManager(null, entityManagerFactory, false, sharedEntityManager, true); + + transactionStrategy.executeInTransaction(() -> { + if (joinTransaction) { + entityManager.joinTransaction(); + } + try { + KeyValueEntry entry = findByKey(entityManager, key); + if (entry == null) { + rc[0] = null; + } else if (entry.isExpired()) { + entityManager.remove(entry); + entityManager.flush(); + rc[0] = null; + } else { + rc[0] = deserialize(entry.getItemValue()); + entityManager.remove(entry); + entityManager.flush(); + } + } catch (Exception ex) { + String contextInfo = String.format(SOMETHING_WENT_WRONG, ex.getMessage()); + throw new PersistenceException(contextInfo, ex); + } finally { + closeEntityManager(entityManager); + } + }); + + LOG.debug("delete {} -> {}", key, rc[0] != null ? "found" : "null"); + return rc[0]; + } + + @Override + @ManagedOperation(description = "Check if key exists") + public boolean contains(String key) { + final Boolean[] rc = new Boolean[1]; + final EntityManager entityManager + = getTargetEntityManager(null, entityManagerFactory, false, sharedEntityManager, true); + + transactionStrategy.executeInTransaction(() -> { + if (joinTransaction) { + entityManager.joinTransaction(); + } + try { + KeyValueEntry entry = findByKey(entityManager, key); + if (entry == null) { + rc[0] = Boolean.FALSE; + } else if (entry.isExpired()) { + entityManager.remove(entry); + entityManager.flush(); + rc[0] = Boolean.FALSE; + } else { + rc[0] = Boolean.TRUE; + } + } catch (Exception ex) { + String contextInfo = String.format(SOMETHING_WENT_WRONG, ex.getMessage()); + throw new PersistenceException(contextInfo, ex); + } finally { + closeEntityManager(entityManager); + } + }); + + LOG.debug("contains {} -> {}", key, rc[0]); + return rc[0]; + } + + @Override + public Set<String> keys() { + final Set<?>[] rc = new Set<?>[1]; + final EntityManager entityManager + = getTargetEntityManager(null, entityManagerFactory, false, sharedEntityManager, true); + + transactionStrategy.executeInTransaction(() -> { + if (joinTransaction) { + entityManager.joinTransaction(); + } + try { + Query query = entityManager.createQuery(QUERY_ALL_KEYS); + query.setParameter(1, System.currentTimeMillis()); + @SuppressWarnings("unchecked") + List<String> resultList = query.getResultList(); + rc[0] = Set.copyOf(resultList); + } catch (Exception ex) { + String contextInfo = String.format(SOMETHING_WENT_WRONG, ex.getMessage()); + throw new PersistenceException(contextInfo, ex); + } finally { + closeEntityManager(entityManager); + } + }); + + LOG.debug("keys -> {} entries", rc[0] != null ? rc[0].size() : 0); + @SuppressWarnings("unchecked") + Set<String> result = (Set<String>) rc[0]; + return result; + } + + @Override + @ManagedOperation(description = "Clear all entries") + public void clear() { + final EntityManager entityManager + = getTargetEntityManager(null, entityManagerFactory, false, sharedEntityManager, true); + + transactionStrategy.executeInTransaction(() -> { + if (joinTransaction) { + entityManager.joinTransaction(); + } + try { + List<?> list = entityManager.createQuery(QUERY_ALL).getResultList(); + if (!list.isEmpty()) { + Iterator<?> it = list.iterator(); + while (it.hasNext()) { + Object item = it.next(); + entityManager.remove(item); + } + entityManager.flush(); + } + } catch (Exception ex) { + String contextInfo = String.format(SOMETHING_WENT_WRONG, ex.getMessage()); + throw new PersistenceException(contextInfo, ex); + } finally { + closeEntityManager(entityManager); + } + }); + + LOG.debug("clear the store {}", KeyValueEntry.class.getName()); + } + + @Override + public Object putIfAbsent(String key, Object value, long ttlMillis) { + final Object[] rc = new Object[1]; + final EntityManager entityManager + = getTargetEntityManager(null, entityManagerFactory, false, sharedEntityManager, true); + + transactionStrategy.executeInTransaction(() -> { + if (joinTransaction) { + entityManager.joinTransaction(); + } + try { + long expiresAt = ttlMillis > 0 ? System.currentTimeMillis() + ttlMillis : 0; + byte[] serializedValue = 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()); + } else if (entry != null) { + // key exists but expired -- update in place + entry.setItemValue(serializedValue); + entry.setExpiresAt(expiresAt); + entityManager.merge(entry); + entityManager.flush(); + rc[0] = null; + } else { + // no entry -- persist new one + entry = new KeyValueEntry(key, serializedValue, expiresAt); + entityManager.persist(entry); + entityManager.flush(); + rc[0] = null; + } + } catch (Exception ex) { + if (isConstraintViolation(ex)) { + // concurrent insert of the same key -- treat as "already present" + LOG.debug("Concurrent insert detected for key: {}", key); + // re-read to return the existing value + try { + KeyValueEntry existing = findByKey(entityManager, key); + rc[0] = existing != null ? deserialize(existing.getItemValue()) : null; + } catch (Exception inner) { + // fall through with null + rc[0] = null; + } + } else { + String contextInfo = String.format(SOMETHING_WENT_WRONG, ex.getMessage()); + throw new PersistenceException(contextInfo, ex); + } + } finally { + closeEntityManager(entityManager); + } + }); + + LOG.debug("putIfAbsent {} -> {}", key, rc[0] != null ? "existing" : "inserted"); + return rc[0]; + } + + @Override + @ManagedAttribute(description = "The number of non-expired entries in the repository") + public int size() { + final int[] rc = new int[1]; + final EntityManager entityManager + = getTargetEntityManager(null, entityManagerFactory, false, sharedEntityManager, true); + + transactionStrategy.executeInTransaction(() -> { + if (joinTransaction) { + entityManager.joinTransaction(); + } + try { + Query query = entityManager.createQuery(QUERY_COUNT_VALID); + query.setParameter(1, System.currentTimeMillis()); + Long count = (Long) query.getSingleResult(); + rc[0] = count.intValue(); + } catch (Exception ex) { + String contextInfo = String.format(SOMETHING_WENT_WRONG, ex.getMessage()); + throw new PersistenceException(contextInfo, ex); + } finally { + closeEntityManager(entityManager); + } + }); + + LOG.debug("size -> {}", rc[0]); + return rc[0]; + } + + // ---- Configuration properties ---- + + public EntityManagerFactory getEntityManagerFactory() { + return entityManagerFactory; + } + + /** + * Sets the JPA EntityManagerFactory to use. + * + * @param entityManagerFactory the entity manager factory + */ + public void setEntityManagerFactory(EntityManagerFactory entityManagerFactory) { + this.entityManagerFactory = entityManagerFactory; + } + + public TransactionStrategy getTransactionStrategy() { + return transactionStrategy; + } + + /** + * Sets the transaction strategy to use. If not set, a {@link DefaultTransactionStrategy} will be created + * automatically. + * + * @param transactionStrategy the transaction strategy + */ + public void setTransactionStrategy(TransactionStrategy transactionStrategy) { + this.transactionStrategy = transactionStrategy; + } + + @ManagedAttribute(description = "Whether to join existing transaction") + public boolean isJoinTransaction() { + return joinTransaction; + } + + /** + * Sets whether to join an existing transaction. Default is {@code true}. + * + * @param joinTransaction whether to join existing transactions + */ + public void setJoinTransaction(boolean joinTransaction) { + this.joinTransaction = joinTransaction; + } + + @ManagedAttribute(description = "Whether to use shared EntityManager") + public boolean isSharedEntityManager() { + return sharedEntityManager; + } + + /** + * Sets whether to use a shared EntityManager. Default is {@code false}. + * + * @param sharedEntityManager whether to use a shared EntityManager + */ + public void setSharedEntityManager(boolean sharedEntityManager) { + this.sharedEntityManager = sharedEntityManager; + } + + // ---- Lifecycle ---- + + @Override + protected void doInit() throws Exception { + if (transactionStrategy == null) { + transactionStrategy = new DefaultTransactionStrategy(null, entityManagerFactory); + } + } + + @Override + protected void doStart() throws Exception { + // noop + } + + @Override + protected void doStop() throws Exception { + // noop + } + + // ---- Private helpers ---- + + private KeyValueEntry findByKey(EntityManager entityManager, String key) { + Query query = entityManager.createQuery(QUERY_BY_KEY); + query.setParameter(1, key); + List<?> list = query.getResultList(); + if (list.isEmpty()) { + return null; + } + return (KeyValueEntry) list.get(0); + } + + 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()) { + entityManager.close(); + } + } catch (Exception e) { + // ignore + } + } + + private static boolean isConstraintViolation(Exception ex) { + Throwable cause = ex; + while (cause != null) { + if (cause instanceof java.sql.SQLIntegrityConstraintViolationException) { + return true; + } + if (cause instanceof jakarta.persistence.EntityExistsException) { + return true; + } + cause = cause.getCause(); + } + return false; + } +} diff --git a/components/camel-jpa/src/main/java/org/apache/camel/processor/keyvalue/jpa/KeyValueEntry.java b/components/camel-jpa/src/main/java/org/apache/camel/processor/keyvalue/jpa/KeyValueEntry.java new file mode 100644 index 000000000000..95517fb6d96d --- /dev/null +++ b/components/camel-jpa/src/main/java/org/apache/camel/processor/keyvalue/jpa/KeyValueEntry.java @@ -0,0 +1,108 @@ +/* + * 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.processor.keyvalue.jpa; + +import java.io.Serial; +import java.io.Serializable; + +import jakarta.persistence.Column; +import jakarta.persistence.Entity; +import jakarta.persistence.Id; +import jakarta.persistence.Lob; +import jakarta.persistence.Table; + +/** + * JPA entity representing a single key-value entry in the {@code CAMEL_KEYVALUE} table. + * <p/> + * The primary key is the {@link #getItemKey() itemKey} (a logical string key). The value is stored as a serialized byte + * array and an optional expiration timestamp controls TTL semantics. + * + * @since 4.23 + */ +@Entity +@Table(name = "CAMEL_KEYVALUE") +public class KeyValueEntry implements Serializable { + + @Serial + private static final long serialVersionUID = 1L; + + private String itemKey; + private byte[] itemValue; + private long expiresAt; + + /** + * Default constructor required by JPA. + */ + public KeyValueEntry() { + } + + /** + * Creates a new key-value entry. + * + * @param itemKey the logical key + * @param itemValue the serialized value + * @param expiresAt the expiration timestamp in epoch milliseconds; {@code 0} means the entry never expires + */ + public KeyValueEntry(String itemKey, byte[] itemValue, long expiresAt) { + this.itemKey = itemKey; + this.itemValue = itemValue; + this.expiresAt = expiresAt; + } + + @Id + @Column(name = "ITEM_KEY", length = 512) + public String getItemKey() { + return itemKey; + } + + public void setItemKey(String itemKey) { + this.itemKey = itemKey; + } + + @Lob + @Column(name = "ITEM_VALUE", nullable = false) + public byte[] getItemValue() { + return itemValue; + } + + public void setItemValue(byte[] itemValue) { + this.itemValue = itemValue; + } + + @Column(name = "EXPIRES_AT", nullable = false) + public long getExpiresAt() { + return expiresAt; + } + + public void setExpiresAt(long expiresAt) { + this.expiresAt = expiresAt; + } + + /** + * Returns {@code true} if this entry has a non-zero expiration time that is in the past. + * + * @return whether the entry is expired + */ + public boolean isExpired() { + return expiresAt > 0 && System.currentTimeMillis() >= expiresAt; + } + + @Override + public String toString() { + return "KeyValueEntry[key=" + itemKey + ", expiresAt=" + expiresAt + "]"; + } +} diff --git a/components/camel-kafka/src/generated/java/org/apache/camel/processor/keyvalue/kafka/KafkaKeyValueRepositoryConfigurer.java b/components/camel-kafka/src/generated/java/org/apache/camel/processor/keyvalue/kafka/KafkaKeyValueRepositoryConfigurer.java new file mode 100644 index 000000000000..3707445d3ad3 --- /dev/null +++ b/components/camel-kafka/src/generated/java/org/apache/camel/processor/keyvalue/kafka/KafkaKeyValueRepositoryConfigurer.java @@ -0,0 +1,90 @@ +/* Generated by camel build tools - do NOT edit this file! */ +package org.apache.camel.processor.keyvalue.kafka; + +import javax.annotation.processing.Generated; +import java.util.Map; + +import org.apache.camel.CamelContext; +import org.apache.camel.spi.ExtendedPropertyConfigurerGetter; +import org.apache.camel.spi.PropertyConfigurerGetter; +import org.apache.camel.spi.ConfigurerStrategy; +import org.apache.camel.spi.GeneratedPropertyConfigurer; +import org.apache.camel.util.CaseInsensitiveMap; +import org.apache.camel.processor.keyvalue.kafka.KafkaKeyValueRepository; + +/** + * Generated by camel build tools - do NOT edit this file! + */ +@Generated("org.apache.camel.maven.packaging.GenerateConfigurerMojo") +@SuppressWarnings("unchecked") +public class KafkaKeyValueRepositoryConfigurer extends org.apache.camel.support.component.PropertyConfigurerSupport implements GeneratedPropertyConfigurer, PropertyConfigurerGetter { + + @Override + public boolean configure(CamelContext camelContext, Object obj, String name, Object value, boolean ignoreCase) { + org.apache.camel.processor.keyvalue.kafka.KafkaKeyValueRepository target = (org.apache.camel.processor.keyvalue.kafka.KafkaKeyValueRepository) obj; + switch (ignoreCase ? name.toLowerCase() : name) { + case "bootstrapservers": + case "bootstrapServers": target.setBootstrapServers(property(camelContext, java.lang.String.class, value)); return true; + case "consumerconfig": + case "consumerConfig": target.setConsumerConfig(property(camelContext, java.util.Properties.class, value)); return true; + case "groupid": + case "groupId": target.setGroupId(property(camelContext, java.lang.String.class, value)); return true; + case "maxcachesize": + case "maxCacheSize": target.setMaxCacheSize(property(camelContext, int.class, value)); return true; + case "polldurationms": + case "pollDurationMs": target.setPollDurationMs(property(camelContext, int.class, value)); return true; + case "producerconfig": + case "producerConfig": target.setProducerConfig(property(camelContext, java.util.Properties.class, value)); return true; + case "startuponly": + case "startupOnly": target.setStartupOnly(property(camelContext, boolean.class, value)); return true; + case "topic": target.setTopic(property(camelContext, java.lang.String.class, value)); return true; + default: return false; + } + } + + @Override + public Class<?> getOptionType(String name, boolean ignoreCase) { + switch (ignoreCase ? name.toLowerCase() : name) { + case "bootstrapservers": + case "bootstrapServers": return java.lang.String.class; + case "consumerconfig": + case "consumerConfig": return java.util.Properties.class; + case "groupid": + case "groupId": return java.lang.String.class; + case "maxcachesize": + case "maxCacheSize": return int.class; + case "polldurationms": + case "pollDurationMs": return int.class; + case "producerconfig": + case "producerConfig": return java.util.Properties.class; + case "startuponly": + case "startupOnly": return boolean.class; + case "topic": return java.lang.String.class; + default: return null; + } + } + + @Override + public Object getOptionValue(Object obj, String name, boolean ignoreCase) { + org.apache.camel.processor.keyvalue.kafka.KafkaKeyValueRepository target = (org.apache.camel.processor.keyvalue.kafka.KafkaKeyValueRepository) obj; + switch (ignoreCase ? name.toLowerCase() : name) { + case "bootstrapservers": + case "bootstrapServers": return target.getBootstrapServers(); + case "consumerconfig": + case "consumerConfig": return target.getConsumerConfig(); + case "groupid": + case "groupId": return target.getGroupId(); + case "maxcachesize": + case "maxCacheSize": return target.getMaxCacheSize(); + case "polldurationms": + case "pollDurationMs": return target.getPollDurationMs(); + case "producerconfig": + case "producerConfig": return target.getProducerConfig(); + case "startuponly": + case "startupOnly": return target.isStartupOnly(); + case "topic": return target.getTopic(); + default: return null; + } + } +} + diff --git a/components/camel-kafka/src/generated/resources/META-INF/services/org/apache/camel/bean.properties b/components/camel-kafka/src/generated/resources/META-INF/services/org/apache/camel/bean.properties index 694e33c8eb14..6a8e4895214d 100644 --- a/components/camel-kafka/src/generated/resources/META-INF/services/org/apache/camel/bean.properties +++ b/components/camel-kafka/src/generated/resources/META-INF/services/org/apache/camel/bean.properties @@ -1,5 +1,5 @@ # Generated by camel build tools - do NOT edit this file! -bean=KafkaIdempotentRepository +bean=KafkaIdempotentRepository KafkaKeyValueRepository groupId=org.apache.camel artifactId=camel-kafka version=4.23.0-SNAPSHOT diff --git a/components/camel-kafka/src/generated/resources/META-INF/services/org/apache/camel/bean/KafkaKeyValueRepository.json b/components/camel-kafka/src/generated/resources/META-INF/services/org/apache/camel/bean/KafkaKeyValueRepository.json new file mode 100644 index 000000000000..dd228413c88e --- /dev/null +++ b/components/camel-kafka/src/generated/resources/META-INF/services/org/apache/camel/bean/KafkaKeyValueRepository.json @@ -0,0 +1,16 @@ +{ + "bean": { + "kind": "bean", + "name": "KafkaKeyValueRepository", + "javaType": "org.apache.camel.processor.keyvalue.kafka.KafkaKeyValueRepository", + "interfaceType": "org.apache.camel.spi.KeyValueRepository", + "title": "Kafka Key Value Repository", + "description": "A Kafka topic-based KeyValueRepository. Uses a local cache backed by a Kafka topic as a changelog. The topic must be unique per logical repository. On startup, the instance consumes the full content of the topic, rebuilding the cache to the latest state.", + "deprecated": false, + "groupId": "org.apache.camel", + "artifactId": "camel-kafka", + "version": "4.23.0-SNAPSHOT", + "properties": { "consumerConfig": { "index": 0, "kind": "property", "displayName": "Consumer Config", "required": false, "type": "object", "javaType": "java.util.Properties", "deprecated": false, "autowired": false, "secret": false, "description": "Custom properties for the Kafka consumer" }, "producerConfig": { "index": 1, "kind": "property", "displayName": "Producer Config", "required": false, "type": "object", "javaType": "java.util.Properties", "deprecated": false, "autowired": f [...] + } +} + diff --git a/components/camel-kafka/src/generated/resources/META-INF/services/org/apache/camel/configurer/org.apache.camel.processor.keyvalue.kafka.KafkaKeyValueRepository b/components/camel-kafka/src/generated/resources/META-INF/services/org/apache/camel/configurer/org.apache.camel.processor.keyvalue.kafka.KafkaKeyValueRepository new file mode 100644 index 000000000000..4c24a898c500 --- /dev/null +++ b/components/camel-kafka/src/generated/resources/META-INF/services/org/apache/camel/configurer/org.apache.camel.processor.keyvalue.kafka.KafkaKeyValueRepository @@ -0,0 +1,2 @@ +# Generated by camel build tools - do NOT edit this file! +class=org.apache.camel.processor.keyvalue.kafka.KafkaKeyValueRepositoryConfigurer 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 new file mode 100644 index 000000000000..6a859661437d --- /dev/null +++ b/components/camel-kafka/src/main/java/org/apache/camel/processor/keyvalue/kafka/KafkaKeyValueRepository.java @@ -0,0 +1,645 @@ +/* + * 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.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; +import java.util.Iterator; +import java.util.List; +import java.util.Map; +import java.util.Properties; +import java.util.Set; +import java.util.concurrent.ExecutionException; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.atomic.AtomicBoolean; +import java.util.concurrent.atomic.AtomicLong; +import java.util.stream.Collectors; + +import org.apache.camel.CamelContext; +import org.apache.camel.CamelContextAware; +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.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.LRUCacheFactory; +import org.apache.camel.support.service.ServiceHelper; +import org.apache.camel.support.service.ServiceSupport; +import org.apache.camel.util.IOHelper; +import org.apache.camel.util.ObjectHelper; +import org.apache.camel.util.StopWatch; +import org.apache.camel.util.StringHelper; +import org.apache.camel.util.TimeUtils; +import org.apache.kafka.clients.consumer.Consumer; +import org.apache.kafka.clients.consumer.ConsumerConfig; +import org.apache.kafka.clients.consumer.ConsumerRecord; +import org.apache.kafka.clients.consumer.ConsumerRecords; +import org.apache.kafka.clients.consumer.KafkaConsumer; +import org.apache.kafka.clients.producer.KafkaProducer; +import org.apache.kafka.clients.producer.Producer; +import org.apache.kafka.clients.producer.ProducerConfig; +import org.apache.kafka.clients.producer.ProducerRecord; +import org.apache.kafka.common.PartitionInfo; +import org.apache.kafka.common.TopicPartition; +import org.apache.kafka.common.errors.WakeupException; +import org.apache.kafka.common.serialization.ByteArrayDeserializer; +import org.apache.kafka.common.serialization.ByteArraySerializer; +import org.apache.kafka.common.serialization.StringDeserializer; +import org.apache.kafka.common.serialization.StringSerializer; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +/** + * A Kafka topic-based implementation of {@link KeyValueRepository}. Uses a local cache backed by a Kafka topic as a + * changelog for durable, distributed key-value storage. + * <p/> + * Each mutation ({@link #put}, {@link #delete}, {@link #clear}) updates the local cache immediately and broadcasts the + * change to the Kafka topic. Other instances consuming the same topic will eventually see the update. On startup, the + * instance consumes the full content of the topic to rebuild the cache to the latest state. + * <p/> + * The topic used must be unique per logical repository. TTL is managed locally via expiration timestamps in the cache; + * expired entries are lazily evicted on access. + * + * @since 4.23 + */ +@Metadata(label = "bean", + description = "A Kafka topic-based KeyValueRepository. Uses a local cache backed by a Kafka topic as a changelog." + + " The topic must be unique per logical repository. On startup, the instance consumes the full content" + + " of the topic, rebuilding the cache to the latest state.", + annotations = { "interfaceName=org.apache.camel.spi.KeyValueRepository" }) +@Configurer(metadataOnly = true) +@ManagedResource(description = "Kafka KeyValueRepository") +public class KafkaKeyValueRepository extends ServiceSupport implements KeyValueRepository, CamelContextAware { + + private static final Logger LOG = LoggerFactory.getLogger(KafkaKeyValueRepository.class); + + private static final int DEFAULT_MAXIMUM_CACHE_SIZE = 1000; + private static final int DEFAULT_POLL_DURATION_MS = 100; + + // Action bytes for the changelog protocol + private static final byte ACTION_PUT = 0; + private static final byte ACTION_DELETE = 1; + private static final byte ACTION_CLEAR = 2; + + private CamelContext camelContext; + private ExecutorService executorService; + private TopicPoller poller; + private final AtomicLong cacheCounter = new AtomicLong(); + + // internal state + private Map<String, CacheEntry> cache; + private Consumer<String, byte[]> consumer; + private Producer<String, byte[]> producer; + + @Metadata(description = "Custom properties for the Kafka consumer") + private Properties consumerConfig; + @Metadata(description = "Custom properties for the Kafka producer") + private Properties producerConfig; + + @Metadata(description = "Sets the name of the Kafka topic used by this repository." + + " Each functionally-separate repository should use a different topic.", + required = true) + private String topic; + @Metadata(description = "The URL for the kafka brokers to use", required = true) + private String bootstrapServers; + @Metadata(description = "A string that uniquely identifies the group of consumer processes to which this consumer belongs.") + private String groupId; + @Metadata(description = "Sets the maximum size of the local key cache.", + defaultValue = "" + DEFAULT_MAXIMUM_CACHE_SIZE) + private int maxCacheSize = DEFAULT_MAXIMUM_CACHE_SIZE; + @Metadata(description = "Sets the poll duration of the Kafka consumer in milliseconds.", + defaultValue = "" + DEFAULT_POLL_DURATION_MS) + private int pollDurationMs = DEFAULT_POLL_DURATION_MS; + @Metadata(description = "Whether to sync on startup only, or to continue syncing while Camel is running.") + private boolean startupOnly; + + public KafkaKeyValueRepository() { + } + + public KafkaKeyValueRepository(String topic, String bootstrapServers) { + this.topic = topic; + this.bootstrapServers = bootstrapServers; + } + + public KafkaKeyValueRepository(String topic, Properties consumerConfig, Properties producerConfig) { + this.topic = topic; + this.consumerConfig = consumerConfig; + this.producerConfig = producerConfig; + } + + // ------------------------------------------------------------------------- + // KeyValueRepository implementation + // ------------------------------------------------------------------------- + + @Override + @ManagedOperation(description = "Get value by key") + public Object get(String key) { + CacheEntry entry = cache.get(key); + if (entry == null) { + return null; + } + if (entry.isExpired()) { + cache.remove(key, entry); + return null; + } + return entry.value; + } + + @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 : 0; + CacheEntry oldEntry = cache.put(key, new CacheEntry(value, expiresAt)); + Object oldValue = (oldEntry != null && !oldEntry.isExpired()) ? oldEntry.value : null; + try { + broadcastPut(key, value, expiresAt); + } catch (Exception e) { + // rollback the cache on broadcast failure + if (oldEntry != null) { + cache.put(key, oldEntry); + } else { + cache.remove(key); + } + throw e; + } + return oldValue; + } + + @Override + @ManagedOperation(description = "Delete a key") + public Object delete(String key) { + CacheEntry oldEntry = cache.remove(key); + Object oldValue = (oldEntry != null && !oldEntry.isExpired()) ? oldEntry.value : null; + broadcastDelete(key); + return oldValue; + } + + @Override + @ManagedOperation(description = "Check if key exists") + public boolean contains(String key) { + CacheEntry entry = cache.get(key); + if (entry == null) { + return false; + } + if (entry.isExpired()) { + cache.remove(key, entry); + return false; + } + return true; + } + + @Override + public Set<String> keys() { + evictExpired(); + return cache.entrySet().stream() + .filter(e -> !e.getValue().isExpired()) + .map(Map.Entry::getKey) + .collect(Collectors.toUnmodifiableSet()); + } + + @Override + @ManagedOperation(description = "Clear all entries") + public void clear() { + cache.clear(); + broadcastClear(); + } + + @Override + public Object putIfAbsent(String key, Object value, long ttlMillis) { + CacheEntry existing = cache.get(key); + if (existing != null && !existing.isExpired()) { + return existing.value; + } + // Remove expired entry if present + if (existing != null) { + cache.remove(key, existing); + } + long expiresAt = ttlMillis > 0 ? System.currentTimeMillis() + ttlMillis : 0; + CacheEntry newEntry = new CacheEntry(value, expiresAt); + CacheEntry prev = cache.putIfAbsent(key, newEntry); + if (prev != null) { + // Another thread beat us + return prev.isExpired() ? null : prev.value; + } + try { + broadcastPut(key, value, expiresAt); + } catch (Exception e) { + cache.remove(key, newEntry); + throw e; + } + return null; + } + + @Override + @ManagedAttribute(description = "The number of entries in the repository") + public int size() { + evictExpired(); + return cache.size(); + } + + // ------------------------------------------------------------------------- + // Broadcast methods + // ------------------------------------------------------------------------- + + private void broadcastPut(String key, Object value, long expiresAt) { + byte[] payload = serializePutAction(value, expiresAt); + broadcastToTopic(key, payload); + } + + private void broadcastDelete(String key) { + broadcastToTopic(key, new byte[] { ACTION_DELETE }); + } + + private void broadcastClear() { + broadcastToTopic(null, new byte[] { ACTION_CLEAR }); + } + + private void broadcastToTopic(String key, byte[] payload) { + try { + LOG.debug("Broadcasting to topic {} for key {}", topic, key); + ObjectHelper.notNull(producer, "producer"); + producer.send(new ProducerRecord<>(topic, key, payload)).get(); // sync send + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + throw new RuntimeCamelException(e); + } catch (ExecutionException e) { + throw new RuntimeCamelException(e); + } + } + + // ------------------------------------------------------------------------- + // Serialization + // ------------------------------------------------------------------------- + + /** + * Serializes a put action: [1 byte action=0][8 bytes expiresAt][serialized Object] + */ + private byte[] serializePutAction(Object value, long expiresAt) { + try { + ByteArrayOutputStream bos = new ByteArrayOutputStream(); + bos.write(ACTION_PUT); + // Write expiresAt as 8 bytes big-endian + ByteBuffer buf = ByteBuffer.allocate(8); + buf.putLong(expiresAt); + bos.write(buf.array()); + // Write serialized value + ObjectOutputStream oos = new ObjectOutputStream(bos); + oos.writeObject(value); + oos.flush(); + return bos.toByteArray(); + } catch (IOException e) { + throw new RuntimeCamelException("Failed to serialize value for Kafka", e); + } + } + + 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); + } + } + + private long deserializeExpiresAt(byte[] data) { + ByteBuffer buf = ByteBuffer.wrap(data, 1, 8); + return buf.getLong(); + } + + // ------------------------------------------------------------------------- + // Cache management + // ------------------------------------------------------------------------- + + private void addToCache(ConsumerRecord<String, byte[]> record) { + cacheCounter.incrementAndGet(); + byte[] data = record.value(); + if (data == null || data.length == 0) { + return; + } + byte action = data[0]; + String key = record.key(); + if (action == ACTION_PUT) { + if (data.length < 10) { + LOG.warn("Malformed put record on topic:{}, partition:{}, offset:{}. Ignoring.", + record.topic(), record.partition(), record.offset()); + return; + } + long expiresAt = deserializeExpiresAt(data); + Object value = deserializeValue(data); + LOG.debug("Adding to cache key:{}", key); + cache.put(key, new CacheEntry(value, expiresAt)); + } else if (action == ACTION_DELETE) { + LOG.debug("Removing from cache key:{}", key); + cache.remove(key); + } else if (action == ACTION_CLEAR) { + LOG.debug("Clearing cache"); + cache.clear(); + } else { + LOG.warn("Unknown action byte:{} on topic:{}, partition:{}, offset:{}. Ignoring.", + action, record.topic(), record.partition(), record.offset()); + } + } + + private void populateCache() { + LOG.debug("Getting partitions of topic {}", topic); + List<PartitionInfo> partitionInfos = consumer.partitionsFor(topic); + Collection<TopicPartition> partitions = partitionInfos.stream() + .map(pi -> new TopicPartition(pi.topic(), pi.partition())) + .toList(); + + LOG.debug("Assigning consumer to partitions {}", partitions); + consumer.assign(partitions); + + LOG.debug("Seeking consumer to beginning of partitions {}", partitions); + consumer.seekToBeginning(partitions); + + Map<TopicPartition, Long> endOffsets = consumer.endOffsets(partitions); + LOG.debug("Consuming records from partitions {} till end offsets {}", partitions, endOffsets); + while (!KafkaConsumerUtil.isReachedOffsets(consumer, endOffsets)) { + ConsumerRecords<String, byte[]> consumerRecords = consumer.poll(Duration.ofMillis(pollDurationMs)); + for (ConsumerRecord<String, byte[]> consumerRecord : consumerRecords) { + addToCache(consumerRecord); + } + } + } + + private void evictExpired() { + Iterator<Map.Entry<String, CacheEntry>> it = cache.entrySet().iterator(); + while (it.hasNext()) { + Map.Entry<String, CacheEntry> entry = it.next(); + if (entry.getValue().isExpired()) { + it.remove(); + } + } + } + + // ------------------------------------------------------------------------- + // Lifecycle + // ------------------------------------------------------------------------- + + @Override + protected void doStart() throws Exception { + ObjectHelper.notNull(camelContext, "camelContext"); + StringHelper.notEmpty(topic, "topic"); + + this.cache = LRUCacheFactory.newLRUCache(maxCacheSize); + + if (consumerConfig == null) { + consumerConfig = new Properties(); + StringHelper.notEmpty(bootstrapServers, "bootstrapServers"); + consumerConfig.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers); + if (groupId != null) { + consumerConfig.put(ConsumerConfig.GROUP_ID_CONFIG, groupId); + } + } + + if (producerConfig == null) { + producerConfig = new Properties(); + StringHelper.notEmpty(bootstrapServers, "bootstrapServers"); + producerConfig.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers); + } + + ObjectHelper.notNull(consumerConfig, "consumerConfig"); + ObjectHelper.notNull(producerConfig, "producerConfig"); + + consumerConfig.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, Boolean.FALSE.toString()); + consumerConfig.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName()); + consumerConfig.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, ByteArrayDeserializer.class.getName()); + + consumer = new KafkaConsumer<>(consumerConfig); + + producerConfig.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName()); + producerConfig.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, ByteArraySerializer.class.getName()); + producerConfig.putIfAbsent(ProducerConfig.ACKS_CONFIG, "1"); + producerConfig.putIfAbsent(ProducerConfig.BATCH_SIZE_CONFIG, "0"); + producer = new KafkaProducer<>(producerConfig); + + poller = new TopicPoller(); + ServiceHelper.startService(poller); + // populate cache on startup + StopWatch watch = new StopWatch(); + LOG.info("Syncing KafkaKeyValueRepository from topic: {} starting", topic); + poller.run(); + LOG.info("Syncing KafkaKeyValueRepository from topic: {} complete: {}", topic, + TimeUtils.printDuration(watch.taken(), true)); + + if (!startupOnly) { + executorService + = camelContext.getExecutorServiceManager().newSingleThreadExecutor(this, "KafkaKeyValueRepositorySync"); + LOG.info("Syncing KafkaKeyValueRepository from topic: {} continuously using background thread", topic); + executorService.submit(poller); + } + } + + @Override + protected void doStop() throws Exception { + ServiceHelper.stopService(poller); + if (consumer != null) { + consumer.wakeup(); + } + if (executorService != null && camelContext != null) { + camelContext.getExecutorServiceManager().shutdownNow(executorService); + executorService = null; + } + IOHelper.close(consumer, "consumer", LOG); + IOHelper.close(producer, "producer", LOG); + LOG.debug("Stopped KafkaKeyValueRepository. Cache counter: {}", cacheCounter.get()); + } + + // ------------------------------------------------------------------------- + // TopicPoller inner class + // ------------------------------------------------------------------------- + + private class TopicPoller extends ServiceSupport implements Runnable { + + private final AtomicBoolean init = new AtomicBoolean(); + + @Override + public void run() { + if (init.compareAndSet(false, true)) { + LOG.debug("TopicPoller populating cache on startup"); + populateCache(); + LOG.debug("TopicPoller populated cache on startup complete"); + return; + } + + LOG.debug("TopicPoller running"); + while (isRunAllowed()) { + try { + ConsumerRecords<String, byte[]> consumerRecords = consumer.poll(Duration.ofMillis(pollDurationMs)); + for (ConsumerRecord<String, byte[]> consumerRecord : consumerRecords) { + addToCache(consumerRecord); + } + } catch (WakeupException e) { + LOG.debug("TopicPoller woken up during shutdown"); + } catch (Exception e) { + LOG.warn("TopicPoller error syncing due to: " + e.getMessage() + ". This exception is ignored.", e); + } + } + LOG.debug("TopicPoller stopping"); + } + } + + // ------------------------------------------------------------------------- + // CacheEntry + // ------------------------------------------------------------------------- + + private static final class CacheEntry { + final Object value; + final long expiresAt; + + CacheEntry(Object value, long expiresAt) { + this.value = value; + this.expiresAt = expiresAt; + } + + boolean isExpired() { + return expiresAt > 0 && System.currentTimeMillis() >= expiresAt; + } + } + + // ------------------------------------------------------------------------- + // CamelContextAware + // ------------------------------------------------------------------------- + + @Override + public void setCamelContext(CamelContext camelContext) { + this.camelContext = camelContext; + } + + @Override + public CamelContext getCamelContext() { + return this.camelContext; + } + + // ------------------------------------------------------------------------- + // Getters & Setters + // ------------------------------------------------------------------------- + + public String getTopic() { + return topic; + } + + /** + * Sets the name of the Kafka topic used by this repository. Each functionally-separate repository should use a + * different topic. + */ + public void setTopic(String topic) { + this.topic = topic; + } + + public String getBootstrapServers() { + return bootstrapServers; + } + + /** + * Sets the bootstrap.servers property on the internal Kafka producer and consumer. + */ + public void setBootstrapServers(String bootstrapServers) { + this.bootstrapServers = bootstrapServers; + } + + public boolean isStartupOnly() { + return startupOnly; + } + + /** + * Whether to sync on startup only, or to continue syncing while Camel is running. + */ + public void setStartupOnly(boolean startupOnly) { + this.startupOnly = startupOnly; + } + + public Properties getProducerConfig() { + return producerConfig; + } + + /** + * Sets the properties that will be used by the Kafka producer. + */ + public void setProducerConfig(Properties producerConfig) { + this.producerConfig = producerConfig; + } + + public Properties getConsumerConfig() { + return consumerConfig; + } + + /** + * Sets the properties that will be used by the Kafka consumer. + */ + public void setConsumerConfig(Properties consumerConfig) { + this.consumerConfig = consumerConfig; + } + + public int getMaxCacheSize() { + return maxCacheSize; + } + + /** + * Sets the maximum size of the local key cache. + */ + public void setMaxCacheSize(int maxCacheSize) { + if (maxCacheSize <= 0) { + throw new IllegalArgumentException("maxCacheSize must be greater than 0, was: " + maxCacheSize); + } + this.maxCacheSize = maxCacheSize; + } + + public int getPollDurationMs() { + return pollDurationMs; + } + + /** + * Sets the poll duration of the Kafka consumer in milliseconds. + */ + public void setPollDurationMs(int pollDurationMs) { + this.pollDurationMs = pollDurationMs; + } + + public String getGroupId() { + return groupId; + } + + /** + * A string that uniquely identifies the group of consumer processes to which this consumer belongs. + */ + public void setGroupId(String groupId) { + this.groupId = groupId; + } + + @ManagedOperation(description = "Number of sync events received from the kafka topic") + public long getCacheCounter() { + return cacheCounter.get(); + } + + @ManagedOperation(description = "Number of elements currently in the cache") + public long getCacheSize() { + return cache != null ? cache.size() : 0; + } +} diff --git a/components/camel-sql/src/generated/java/org/apache/camel/processor/keyvalue/jdbc/JdbcKeyValueRepositoryConfigurer.java b/components/camel-sql/src/generated/java/org/apache/camel/processor/keyvalue/jdbc/JdbcKeyValueRepositoryConfigurer.java new file mode 100644 index 000000000000..91704ed283be --- /dev/null +++ b/components/camel-sql/src/generated/java/org/apache/camel/processor/keyvalue/jdbc/JdbcKeyValueRepositoryConfigurer.java @@ -0,0 +1,117 @@ +/* Generated by camel build tools - do NOT edit this file! */ +package org.apache.camel.processor.keyvalue.jdbc; + +import javax.annotation.processing.Generated; +import java.util.Map; + +import org.apache.camel.CamelContext; +import org.apache.camel.spi.ExtendedPropertyConfigurerGetter; +import org.apache.camel.spi.PropertyConfigurerGetter; +import org.apache.camel.spi.ConfigurerStrategy; +import org.apache.camel.spi.GeneratedPropertyConfigurer; +import org.apache.camel.util.CaseInsensitiveMap; +import org.apache.camel.processor.keyvalue.jdbc.JdbcKeyValueRepository; + +/** + * Generated by camel build tools - do NOT edit this file! + */ +@Generated("org.apache.camel.maven.packaging.GenerateConfigurerMojo") +@SuppressWarnings("unchecked") +public class JdbcKeyValueRepositoryConfigurer extends org.apache.camel.support.component.PropertyConfigurerSupport implements GeneratedPropertyConfigurer, PropertyConfigurerGetter { + + @Override + public boolean configure(CamelContext camelContext, Object obj, String name, Object value, boolean ignoreCase) { + org.apache.camel.processor.keyvalue.jdbc.JdbcKeyValueRepository target = (org.apache.camel.processor.keyvalue.jdbc.JdbcKeyValueRepository) obj; + switch (ignoreCase ? name.toLowerCase() : name) { + case "clearstring": + case "clearString": target.setClearString(property(camelContext, java.lang.String.class, value)); return true; + case "createstring": + case "createString": target.setCreateString(property(camelContext, java.lang.String.class, value)); return true; + case "createtableifnotexists": + case "createTableIfNotExists": target.setCreateTableIfNotExists(property(camelContext, boolean.class, value)); return true; + case "deleteexpiredstring": + case "deleteExpiredString": target.setDeleteExpiredString(property(camelContext, java.lang.String.class, value)); return true; + case "deletestring": + case "deleteString": target.setDeleteString(property(camelContext, java.lang.String.class, value)); return true; + case "insertstring": + case "insertString": target.setInsertString(property(camelContext, java.lang.String.class, value)); return true; + case "jdbctemplate": + case "jdbcTemplate": target.setJdbcTemplate(property(camelContext, org.springframework.jdbc.core.JdbcTemplate.class, value)); return true; + case "selectkeysstring": + case "selectKeysString": target.setSelectKeysString(property(camelContext, java.lang.String.class, value)); return true; + case "selectstring": + case "selectString": target.setSelectString(property(camelContext, java.lang.String.class, value)); return true; + case "tableexistsstring": + case "tableExistsString": target.setTableExistsString(property(camelContext, java.lang.String.class, value)); return true; + case "tablename": + case "tableName": target.setTableName(property(camelContext, java.lang.String.class, value)); return true; + case "transactiontemplate": + case "transactionTemplate": target.setTransactionTemplate(property(camelContext, org.springframework.transaction.support.TransactionTemplate.class, value)); return true; + default: return false; + } + } + + @Override + public Class<?> getOptionType(String name, boolean ignoreCase) { + switch (ignoreCase ? name.toLowerCase() : name) { + case "clearstring": + case "clearString": return java.lang.String.class; + case "createstring": + case "createString": return java.lang.String.class; + case "createtableifnotexists": + case "createTableIfNotExists": return boolean.class; + case "deleteexpiredstring": + case "deleteExpiredString": return java.lang.String.class; + case "deletestring": + case "deleteString": return java.lang.String.class; + case "insertstring": + case "insertString": return java.lang.String.class; + case "jdbctemplate": + case "jdbcTemplate": return org.springframework.jdbc.core.JdbcTemplate.class; + case "selectkeysstring": + case "selectKeysString": return java.lang.String.class; + case "selectstring": + case "selectString": return java.lang.String.class; + case "tableexistsstring": + case "tableExistsString": return java.lang.String.class; + case "tablename": + case "tableName": return java.lang.String.class; + case "transactiontemplate": + case "transactionTemplate": return org.springframework.transaction.support.TransactionTemplate.class; + default: return null; + } + } + + @Override + public Object getOptionValue(Object obj, String name, boolean ignoreCase) { + org.apache.camel.processor.keyvalue.jdbc.JdbcKeyValueRepository target = (org.apache.camel.processor.keyvalue.jdbc.JdbcKeyValueRepository) obj; + switch (ignoreCase ? name.toLowerCase() : name) { + case "clearstring": + case "clearString": return target.getClearString(); + case "createstring": + case "createString": return target.getCreateString(); + case "createtableifnotexists": + case "createTableIfNotExists": return target.isCreateTableIfNotExists(); + case "deleteexpiredstring": + case "deleteExpiredString": return target.getDeleteExpiredString(); + case "deletestring": + case "deleteString": return target.getDeleteString(); + case "insertstring": + case "insertString": return target.getInsertString(); + case "jdbctemplate": + case "jdbcTemplate": return target.getJdbcTemplate(); + case "selectkeysstring": + case "selectKeysString": return target.getSelectKeysString(); + case "selectstring": + case "selectString": return target.getSelectString(); + case "tableexistsstring": + case "tableExistsString": return target.getTableExistsString(); + case "tablename": + case "tableName": return target.getTableName(); + case "transactiontemplate": + case "transactionTemplate": return target.getTransactionTemplate(); + default: return null; + } + } +} + diff --git a/components/camel-sql/src/generated/resources/META-INF/services/org/apache/camel/bean.properties b/components/camel-sql/src/generated/resources/META-INF/services/org/apache/camel/bean.properties index 6750f51f35ec..aa8d87282f47 100644 --- a/components/camel-sql/src/generated/resources/META-INF/services/org/apache/camel/bean.properties +++ b/components/camel-sql/src/generated/resources/META-INF/services/org/apache/camel/bean.properties @@ -1,5 +1,5 @@ # Generated by camel build tools - do NOT edit this file! -bean=JdbcAggregationRepository JdbcMessageIdRepository +bean=JdbcAggregationRepository JdbcKeyValueRepository JdbcMessageIdRepository groupId=org.apache.camel artifactId=camel-sql version=4.23.0-SNAPSHOT diff --git a/components/camel-sql/src/generated/resources/META-INF/services/org/apache/camel/bean/JdbcKeyValueRepository.json b/components/camel-sql/src/generated/resources/META-INF/services/org/apache/camel/bean/JdbcKeyValueRepository.json new file mode 100644 index 000000000000..a38b669e8cca --- /dev/null +++ b/components/camel-sql/src/generated/resources/META-INF/services/org/apache/camel/bean/JdbcKeyValueRepository.json @@ -0,0 +1,16 @@ +{ + "bean": { + "kind": "bean", + "name": "JdbcKeyValueRepository", + "javaType": "org.apache.camel.processor.keyvalue.jdbc.JdbcKeyValueRepository", + "interfaceType": "org.apache.camel.spi.KeyValueRepository", + "title": "Jdbc Key Value Repository", + "description": "A JDBC-based KeyValueRepository that stores entries in a database table.", + "deprecated": false, + "groupId": "org.apache.camel", + "artifactId": "camel-sql", + "version": "4.23.0-SNAPSHOT", + "properties": { "jdbcTemplate": { "index": 0, "kind": "property", "displayName": "Jdbc Template", "required": true, "type": "object", "javaType": "org.springframework.jdbc.core.JdbcTemplate", "deprecated": false, "autowired": false, "secret": false, "description": "The Spring JdbcTemplate to use for connecting to the database" }, "transactionTemplate": { "index": 1, "kind": "property", "displayName": "Transaction Template", "required": true, "type": "object", "javaType": "org.springf [...] + } +} + diff --git a/components/camel-sql/src/generated/resources/META-INF/services/org/apache/camel/configurer/org.apache.camel.processor.keyvalue.jdbc.JdbcKeyValueRepository b/components/camel-sql/src/generated/resources/META-INF/services/org/apache/camel/configurer/org.apache.camel.processor.keyvalue.jdbc.JdbcKeyValueRepository new file mode 100644 index 000000000000..b83195ec71bb --- /dev/null +++ b/components/camel-sql/src/generated/resources/META-INF/services/org/apache/camel/configurer/org.apache.camel.processor.keyvalue.jdbc.JdbcKeyValueRepository @@ -0,0 +1,2 @@ +# Generated by camel build tools - do NOT edit this file! +class=org.apache.camel.processor.keyvalue.jdbc.JdbcKeyValueRepositoryConfigurer 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 new file mode 100644 index 000000000000..9b7cc195fc8b --- /dev/null +++ b/components/camel-sql/src/main/java/org/apache/camel/processor/keyvalue/jdbc/JdbcKeyValueRepository.java @@ -0,0 +1,470 @@ +/* + * 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.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.util.Collections; +import java.util.LinkedHashSet; +import java.util.List; +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.service.ServiceSupport; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; +import org.springframework.dao.DataAccessException; +import org.springframework.dao.DuplicateKeyException; +import org.springframework.dao.EmptyResultDataAccessException; +import org.springframework.jdbc.core.JdbcTemplate; +import org.springframework.jdbc.datasource.DataSourceTransactionManager; +import org.springframework.transaction.TransactionDefinition; +import org.springframework.transaction.support.TransactionTemplate; + +/** + * A JDBC-based implementation of {@link KeyValueRepository} that stores entries in a database table. + * <p/> + * Values are serialized using Java object serialization and stored as BLOB. Each entry may optionally have a + * time-to-live (TTL); expired entries are lazily evicted on access and during {@link #keys()} scans. + * <p/> + * The table is created automatically on startup if it does not already exist (controlled by + * {@link #setCreateTableIfNotExists(boolean)}). + * + * @since 4.23 + */ +@Metadata(label = "bean", + description = "A JDBC-based KeyValueRepository that stores entries in a database table.", + annotations = { "interfaceName=org.apache.camel.spi.KeyValueRepository" }) +@Configurer(metadataOnly = true) +@ManagedResource(description = "JDBC based key-value repository") +public class JdbcKeyValueRepository extends ServiceSupport implements KeyValueRepository { + + protected static final String DEFAULT_TABLENAME = "CAMEL_KEYVALUE"; + protected static final String DEFAULT_TABLE_EXISTS_STRING = "SELECT 1 FROM CAMEL_KEYVALUE WHERE 1 = 0"; + protected static final String DEFAULT_CREATE_STRING + = "CREATE TABLE CAMEL_KEYVALUE (ITEM_KEY VARCHAR(512) NOT NULL, ITEM_VALUE BLOB NOT NULL, " + + "EXPIRES_AT BIGINT NOT NULL DEFAULT 0, PRIMARY KEY (ITEM_KEY))"; + protected static final String DEFAULT_SELECT_STRING + = "SELECT ITEM_VALUE, EXPIRES_AT FROM CAMEL_KEYVALUE WHERE ITEM_KEY = ?"; + protected static final String DEFAULT_INSERT_STRING + = "INSERT INTO CAMEL_KEYVALUE (ITEM_KEY, ITEM_VALUE, EXPIRES_AT) VALUES (?, ?, ?)"; + protected static final String DEFAULT_DELETE_STRING = "DELETE FROM CAMEL_KEYVALUE WHERE ITEM_KEY = ?"; + protected static final String DEFAULT_CLEAR_STRING = "DELETE FROM CAMEL_KEYVALUE"; + protected static final String DEFAULT_SELECT_KEYS_STRING + = "SELECT ITEM_KEY FROM CAMEL_KEYVALUE WHERE EXPIRES_AT = 0 OR EXPIRES_AT > ?"; + protected static final String DEFAULT_DELETE_EXPIRED_STRING + = "DELETE FROM CAMEL_KEYVALUE WHERE EXPIRES_AT > 0 AND EXPIRES_AT <= ?"; + + private static final Logger LOG = LoggerFactory.getLogger(JdbcKeyValueRepository.class); + + @Metadata(description = "The Spring JdbcTemplate to use for connecting to the database", required = true) + private JdbcTemplate jdbcTemplate; + @Metadata(description = "The Spring TransactionTemplate to use for connecting to the database", required = true) + private TransactionTemplate transactionTemplate; + private DataSource dataSource; + + @Metadata(description = "The name of the table to use in the database", defaultValue = "CAMEL_KEYVALUE") + private String tableName; + @Metadata(description = "Whether to create the table in the database if none exists on startup", defaultValue = "true") + private boolean createTableIfNotExists = true; + + @Metadata(label = "advanced", description = "SQL query to use for checking if table exists") + private String tableExistsString = DEFAULT_TABLE_EXISTS_STRING; + @Metadata(label = "advanced", description = "SQL query to use for creating table") + private String createString = DEFAULT_CREATE_STRING; + @Metadata(label = "advanced", description = "SQL query to use for selecting a value by key") + private String selectString = DEFAULT_SELECT_STRING; + @Metadata(label = "advanced", description = "SQL query to use for inserting a new entry") + private String insertString = DEFAULT_INSERT_STRING; + @Metadata(label = "advanced", description = "SQL query to use for deleting an entry by key") + private String deleteString = DEFAULT_DELETE_STRING; + @Metadata(label = "advanced", description = "SQL query to delete all entries from the table") + private String clearString = DEFAULT_CLEAR_STRING; + @Metadata(label = "advanced", description = "SQL query to use for selecting all non-expired keys") + private String selectKeysString = DEFAULT_SELECT_KEYS_STRING; + @Metadata(label = "advanced", description = "SQL query to use for deleting expired entries") + private String deleteExpiredString = DEFAULT_DELETE_EXPIRED_STRING; + + /** + * Creates a new JDBC key-value repository. A {@link DataSource} or {@link JdbcTemplate} must be set before + * initialization. + */ + public JdbcKeyValueRepository() { + } + + /** + * Creates a new JDBC key-value repository using the given data source. A {@link JdbcTemplate} and + * {@link TransactionTemplate} will be created automatically during initialization. + * + * @param dataSource the data source to use + */ + public JdbcKeyValueRepository(DataSource dataSource) { + this.dataSource = dataSource; + } + + /** + * Creates a new JDBC key-value repository using the given JDBC template and transaction template. + * + * @param jdbcTemplate the JDBC template for database access + * @param transactionTemplate the transaction template for transactional operations + */ + public JdbcKeyValueRepository(JdbcTemplate jdbcTemplate, TransactionTemplate transactionTemplate) { + this.jdbcTemplate = jdbcTemplate; + this.transactionTemplate = transactionTemplate; + } + + /** + * Creates a {@link TransactionTemplate} from the given data source with {@code PROPAGATION_REQUIRED}. + * + * @param dataSource the data source to create the transaction template from + * @return a configured transaction template + */ + protected static TransactionTemplate createTransactionTemplate(DataSource dataSource) { + TransactionTemplate transactionTemplate = new TransactionTemplate(); + transactionTemplate.setTransactionManager(new DataSourceTransactionManager(dataSource)); + transactionTemplate.setPropagationBehavior(TransactionDefinition.PROPAGATION_REQUIRED); + return transactionTemplate; + } + + @Override + protected void doInit() throws Exception { + super.doInit(); + + if (dataSource != null && jdbcTemplate == null) { + jdbcTemplate = new JdbcTemplate(dataSource); + jdbcTemplate.afterPropertiesSet(); + } + if (dataSource != null && transactionTemplate == null) { + transactionTemplate = createTransactionTemplate(dataSource); + } + + if (tableName != null) { + // update query strings from default table name to the custom table name + tableExistsString = DEFAULT_TABLE_EXISTS_STRING.replace(DEFAULT_TABLENAME, tableName); + createString = DEFAULT_CREATE_STRING.replace(DEFAULT_TABLENAME, tableName); + selectString = DEFAULT_SELECT_STRING.replace(DEFAULT_TABLENAME, tableName); + insertString = DEFAULT_INSERT_STRING.replace(DEFAULT_TABLENAME, tableName); + deleteString = DEFAULT_DELETE_STRING.replace(DEFAULT_TABLENAME, tableName); + clearString = DEFAULT_CLEAR_STRING.replace(DEFAULT_TABLENAME, tableName); + selectKeysString = DEFAULT_SELECT_KEYS_STRING.replace(DEFAULT_TABLENAME, tableName); + deleteExpiredString = DEFAULT_DELETE_EXPIRED_STRING.replace(DEFAULT_TABLENAME, tableName); + } + } + + @Override + protected void doStart() throws Exception { + super.doStart(); + + boolean tableExists = transactionTemplate.execute(status -> { + try { + // we will receive an exception if the table doesn't exist or we cannot access it + jdbcTemplate.execute(getTableExistsString()); + LOG.debug("Expected table for JdbcKeyValueRepository exists"); + return true; + } catch (DataAccessException e) { + LOG.debug("Expected table for JdbcKeyValueRepository does not exist"); + return false; + } + }); + + if (!tableExists && createTableIfNotExists) { + transactionTemplate.executeWithoutResult(status -> { + try { + LOG.debug("Creating table for JdbcKeyValueRepository because it doesn't exist..."); + jdbcTemplate.execute(getCreateString()); + LOG.info("Table created with query '{}'", getCreateString()); + } catch (DataAccessException dae) { + LOG.error( + "Can't create table for JdbcKeyValueRepository with query '{}' because of: {}. " + + "This may be a permissions problem. Please create this table and try again.", + getCreateString(), dae.getMessage()); + throw dae; + } + }); + } + } + + @Override + protected void doStop() throws Exception { + // noop + } + + @Override + @ManagedOperation(description = "Get value by key") + public Object get(String key) { + return transactionTemplate.execute(status -> doGet(key)); + } + + @Override + @ManagedOperation(description = "Put a key-value pair with optional TTL") + public Object put(String key, Object value, long ttlMillis) { + return transactionTemplate.execute(status -> { + Object oldValue = doGet(key); + // delete any existing row (whether expired or not) + jdbcTemplate.update(getDeleteString(), key); + // insert the new row + long expiresAt = ttlMillis > 0 ? System.currentTimeMillis() + ttlMillis : 0; + jdbcTemplate.update(getInsertString(), key, serialize(value), expiresAt); + return oldValue; + }); + } + + @Override + @ManagedOperation(description = "Delete a key") + public Object delete(String key) { + return transactionTemplate.execute(status -> { + Object oldValue = doGet(key); + jdbcTemplate.update(getDeleteString(), key); + return oldValue; + }); + } + + @Override + @ManagedOperation(description = "Check if key exists") + public boolean contains(String key) { + Boolean result = transactionTemplate.execute(status -> doGet(key) != null); + return result != null && result; + } + + @Override + public Set<String> keys() { + return transactionTemplate.execute(status -> { + // first delete expired entries + long now = System.currentTimeMillis(); + jdbcTemplate.update(getDeleteExpiredString(), now); + // then select all non-expired keys + List<String> keyList = jdbcTemplate.queryForList(getSelectKeysString(), String.class, now); + return Collections.unmodifiableSet(new LinkedHashSet<>(keyList)); + }); + } + + @Override + @ManagedOperation(description = "Clear all entries") + public void clear() { + transactionTemplate.executeWithoutResult(status -> jdbcTemplate.update(getClearString())); + } + + @Override + public Object putIfAbsent(String key, Object value, long ttlMillis) { + return transactionTemplate.execute(status -> { + // check if a non-expired entry already exists + Object existing = doGet(key); + if (existing != null) { + return existing; + } + // attempt to insert + long expiresAt = ttlMillis > 0 ? System.currentTimeMillis() + ttlMillis : 0; + try { + jdbcTemplate.update(getInsertString(), key, serialize(value), expiresAt); + return null; + } catch (DuplicateKeyException e) { + // concurrent insert race -- another thread/node won + LOG.debug("Concurrent insert race for key '{}' -- another thread won, treating as existing", key); + status.setRollbackOnly(); + // re-read the value that the other thread inserted + Object concurrentValue = doGet(key); + return concurrentValue; + } + }); + } + + @Override + @ManagedAttribute(description = "The number of entries in the repository") + public int size() { + return keys().size(); + } + + /** + * Internal get that reads the value for the given key within the current transaction. If the entry has expired, it + * is deleted and {@code null} is returned. + * + * @param key the key to look up + * @return the deserialized value, or {@code null} if not found or expired + */ + private Object doGet(String key) { + try { + return jdbcTemplate.queryForObject(getSelectString(), (rs, rowNum) -> { + byte[] bytes = rs.getBytes(1); + long expiresAt = rs.getLong(2); + if (expiresAt > 0 && System.currentTimeMillis() >= expiresAt) { + // entry has expired -- delete it + jdbcTemplate.update(getDeleteString(), key); + return null; + } + return deserialize(bytes); + }, key); + } catch (EmptyResultDataAccessException e) { + return null; + } + } + + /** + * 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") + public String getTableName() { + return tableName; + } + + /** + * To use a custom table name instead of the default name: CAMEL_KEYVALUE + */ + public void setTableName(String tableName) { + this.tableName = tableName; + } + + @ManagedAttribute(description = "Whether to create the table if it does not exist on startup") + public boolean isCreateTableIfNotExists() { + return createTableIfNotExists; + } + + public void setCreateTableIfNotExists(boolean createTableIfNotExists) { + this.createTableIfNotExists = createTableIfNotExists; + } + + public JdbcTemplate getJdbcTemplate() { + return jdbcTemplate; + } + + public void setJdbcTemplate(JdbcTemplate jdbcTemplate) { + this.jdbcTemplate = jdbcTemplate; + } + + public TransactionTemplate getTransactionTemplate() { + return transactionTemplate; + } + + public void setTransactionTemplate(TransactionTemplate transactionTemplate) { + this.transactionTemplate = transactionTemplate; + } + + public DataSource getDataSource() { + return dataSource; + } + + public void setDataSource(DataSource dataSource) { + this.dataSource = dataSource; + } + + public String getTableExistsString() { + return tableExistsString; + } + + public void setTableExistsString(String tableExistsString) { + this.tableExistsString = tableExistsString; + } + + public String getCreateString() { + return createString; + } + + public void setCreateString(String createString) { + this.createString = createString; + } + + public String getSelectString() { + return selectString; + } + + public void setSelectString(String selectString) { + this.selectString = selectString; + } + + public String getInsertString() { + return insertString; + } + + public void setInsertString(String insertString) { + this.insertString = insertString; + } + + public String getDeleteString() { + return deleteString; + } + + public void setDeleteString(String deleteString) { + this.deleteString = deleteString; + } + + public String getClearString() { + return clearString; + } + + public void setClearString(String clearString) { + this.clearString = clearString; + } + + public String getSelectKeysString() { + return selectKeysString; + } + + public void setSelectKeysString(String selectKeysString) { + this.selectKeysString = selectKeysString; + } + + public String getDeleteExpiredString() { + return deleteExpiredString; + } + + public void setDeleteExpiredString(String deleteExpiredString) { + this.deleteExpiredString = deleteExpiredString; + } +} diff --git a/components/camel-sql/src/test/java/org/apache/camel/processor/keyvalue/jdbc/JdbcKeyValueRepositoryTest.java b/components/camel-sql/src/test/java/org/apache/camel/processor/keyvalue/jdbc/JdbcKeyValueRepositoryTest.java new file mode 100644 index 000000000000..fd1f2c89c086 --- /dev/null +++ b/components/camel-sql/src/test/java/org/apache/camel/processor/keyvalue/jdbc/JdbcKeyValueRepositoryTest.java @@ -0,0 +1,243 @@ +/* + * 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.processor.keyvalue.jdbc; + +import java.util.Set; +import java.util.concurrent.TimeUnit; + +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; +import org.springframework.jdbc.datasource.embedded.EmbeddedDatabase; +import org.springframework.jdbc.datasource.embedded.EmbeddedDatabaseBuilder; +import org.springframework.jdbc.datasource.embedded.EmbeddedDatabaseType; + +import static org.awaitility.Awaitility.await; +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertNull; +import static org.junit.jupiter.api.Assertions.assertTrue; + +class JdbcKeyValueRepositoryTest { + + private JdbcKeyValueRepository repository; + private EmbeddedDatabase dataSource; + + @BeforeEach + void setUp() throws Exception { + dataSource = new EmbeddedDatabaseBuilder() + .setType(EmbeddedDatabaseType.H2) + .build(); + repository = new JdbcKeyValueRepository(dataSource); + repository.init(); + repository.start(); + } + + @AfterEach + void tearDown() throws Exception { + repository.stop(); + dataSource.shutdown(); + } + + @Test + void testPutAndGet() { + repository.put("key1", "value1", 0); + assertEquals("value1", repository.get("key1")); + } + + @Test + void testGetMissingKeyReturnsNull() { + assertNull(repository.get("nonexistent")); + } + + @Test + void testPutOverwritesExistingValue() { + repository.put("key1", "value1", 0); + repository.put("key1", "value2", 0); + assertEquals("value2", repository.get("key1")); + } + + @Test + void testPutReturnsOldValue() { + repository.put("key1", "value1", 0); + Object old = repository.put("key1", "value2", 0); + assertEquals("value1", old); + } + + @Test + void testPutReturnsNullForNewKey() { + Object old = repository.put("key1", "value1", 0); + assertNull(old); + } + + @Test + void testDelete() { + repository.put("key1", "value1", 0); + Object deleted = repository.delete("key1"); + assertEquals("value1", deleted); + assertNull(repository.get("key1")); + } + + @Test + void testDeleteMissingKeyReturnsNull() { + assertNull(repository.delete("nonexistent")); + } + + @Test + void testContains() { + repository.put("key1", "value1", 0); + assertTrue(repository.contains("key1")); + assertFalse(repository.contains("nonexistent")); + } + + @Test + void testKeys() { + repository.put("key1", "value1", 0); + repository.put("key2", "value2", 0); + repository.put("key3", "value3", 0); + + Set<String> keys = repository.keys(); + assertEquals(3, keys.size()); + assertTrue(keys.contains("key1")); + assertTrue(keys.contains("key2")); + assertTrue(keys.contains("key3")); + } + + @Test + void testKeysEmpty() { + assertTrue(repository.keys().isEmpty()); + } + + @Test + void testClear() { + repository.put("key1", "value1", 0); + repository.put("key2", "value2", 0); + repository.clear(); + assertEquals(0, repository.size()); + assertNull(repository.get("key1")); + assertNull(repository.get("key2")); + } + + @Test + void testSize() { + assertEquals(0, repository.size()); + repository.put("key1", "value1", 0); + assertEquals(1, repository.size()); + repository.put("key2", "value2", 0); + assertEquals(2, repository.size()); + repository.delete("key1"); + assertEquals(1, repository.size()); + } + + @Test + void testPutIfAbsentNewKey() { + Object result = repository.putIfAbsent("key1", "value1", 0); + assertNull(result); + assertEquals("value1", repository.get("key1")); + } + + @Test + void testPutIfAbsentExistingKey() { + repository.put("key1", "value1", 0); + Object result = repository.putIfAbsent("key1", "value2", 0); + assertEquals("value1", result); + assertEquals("value1", repository.get("key1")); + } + + @Test + void testTtlExpiration() { + repository.put("key1", "value1", 50); + assertEquals("value1", repository.get("key1")); + assertTrue(repository.contains("key1")); + + await().atMost(500, TimeUnit.MILLISECONDS) + .untilAsserted(() -> { + assertNull(repository.get("key1")); + assertFalse(repository.contains("key1")); + }); + } + + @Test + void testTtlExpirationOnKeys() { + repository.put("key1", "value1", 50); + repository.put("key2", "value2", 0); // no expiration + + await().atMost(500, TimeUnit.MILLISECONDS) + .untilAsserted(() -> { + Set<String> keys = repository.keys(); + assertEquals(1, keys.size()); + assertTrue(keys.contains("key2")); + }); + } + + @Test + void testTtlExpirationOnDelete() { + repository.put("key1", "value1", 50); + + await().atMost(500, TimeUnit.MILLISECONDS) + .untilAsserted(() -> assertNull(repository.delete("key1"))); + } + + @Test + void testPutIfAbsentWithExpiredEntry() { + repository.put("key1", "value1", 50); + + await().atMost(500, TimeUnit.MILLISECONDS) + .untilAsserted(() -> { + Object result = repository.putIfAbsent("key1", "value2", 0); + assertNull(result); + assertEquals("value2", repository.get("key1")); + }); + } + + @Test + void testNoTtlWithZero() { + repository.put("key1", "value1", 0); + assertEquals("value1", repository.get("key1")); + assertTrue(repository.contains("key1")); + } + + @Test + void testNoTtlWithNegative() { + repository.put("key1", "value1", -1); + assertEquals("value1", repository.get("key1")); + assertTrue(repository.contains("key1")); + } + + @Test + void testStoresDifferentValueTypes() { + repository.put("string", "hello", 0); + repository.put("integer", 42, 0); + repository.put("boolean", Boolean.TRUE, 0); + + assertEquals("hello", repository.get("string")); + assertEquals(42, repository.get("integer")); + assertEquals(Boolean.TRUE, repository.get("boolean")); + } + + @Test + void testAutoCreateTable() throws Exception { + repository.stop(); + repository = new JdbcKeyValueRepository(dataSource); + repository.setTableName("CUSTOM_KV_TABLE"); + repository.init(); + repository.start(); + + repository.put("key1", "value1", 0); + assertEquals("value1", repository.get("key1")); + } +}
