davsclaus commented on code in PR #25993: URL: https://github.com/apache/camel/pull/25993#discussion_r3913781628
########## components/camel-kafka/src/main/java/org/apache/camel/processor/keyvalue/kafka/KafkaKeyValueRepository.java: ########## @@ -0,0 +1,642 @@ +/* + * 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.ByteArrayOutputStream; +import java.io.IOException; +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.KeyValueRepositoryHelper; +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, Duration ttl) { + long expiresAt = toExpiresAt(ttl); + 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); Review Comment: `put()` and `putIfAbsent()` restore the local cache if the synchronous broadcast fails, but `delete()` (here) and `clear()` (below) mutate the cache and then broadcast with no rollback. If the broadcast throws, local state diverges from what peers will replay from the topic. Consider mirroring the try/catch rollback used in `put()` — or, if the divergence is intentional (local removal is still a valid state), a short comment would help future readers. ########## components/camel-kafka/src/main/java/org/apache/camel/processor/keyvalue/kafka/KafkaKeyValueRepository.java: ########## @@ -0,0 +1,642 @@ +/* + * 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.ByteArrayOutputStream; +import java.io.IOException; +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.KeyValueRepositoryHelper; +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, Duration ttl) { + long expiresAt = toExpiresAt(ttl); + 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, Duration ttl) { + 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 = toExpiresAt(ttl); + 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 + bos.write(KeyValueRepositoryHelper.serialize(value)); + return bos.toByteArray(); + } catch (IOException e) { + throw new RuntimeCamelException("Failed to serialize value for Kafka", e); + } + } + + private Object deserializeValue(byte[] data) { + // Value starts at offset 9 (1 byte action + 8 bytes expiresAt) + return KeyValueRepositoryHelper.deserialize(data, 9, data.length - 9); + } + + 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 static long toExpiresAt(Duration ttl) { + if (ttl == null || ttl.isZero() || ttl.isNegative()) { + return 0; + } + return System.currentTimeMillis() + ttl.toMillis(); + } + + 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); Review Comment: Reads only ever consult this in-memory LRU cache. When the number of live keys exceeds `maxCacheSize` (default 1000), still-valid entries are silently evicted from memory and `get()`/`contains()` will report them as absent even though the value persists in the compacted topic. This matches the `KafkaIdempotentRepository` trade-off, but as a general-purpose `KeyValueRepository` (backing the Aggregator and Cache EIP) the silent under-reporting of live state is more surprising. Worth calling out prominently in the doc page so operators size `maxCacheSize` to the expected key cardinality. ########## components/camel-sql/src/main/java/org/apache/camel/processor/keyvalue/jdbc/JdbcKeyValueRepository.java: ########## @@ -0,0 +1,439 @@ +/* + * 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.time.Duration; +import java.util.Collections; +import java.util.LinkedHashSet; +import java.util.List; +import java.util.Set; + +import javax.sql.DataSource; + +import org.apache.camel.api.management.ManagedAttribute; +import org.apache.camel.api.management.ManagedOperation; +import org.apache.camel.api.management.ManagedResource; +import org.apache.camel.spi.Configurer; +import org.apache.camel.spi.KeyValueRepository; +import org.apache.camel.spi.Metadata; +import org.apache.camel.support.KeyValueRepositoryHelper; +import org.apache.camel.support.service.ServiceSupport; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; +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, Duration ttl) { + 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 = toExpiresAt(ttl); + jdbcTemplate.update(getInsertString(), key, KeyValueRepositoryHelper.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, Duration ttl) { + 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 = toExpiresAt(ttl); + try { + jdbcTemplate.update(getInsertString(), key, KeyValueRepositoryHelper.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); Review Comment: Issuing a `DELETE` from inside the `queryForObject` RowMapper (while the ResultSet for the same query is still open on the same connection) works on H2, but some JDBC drivers dislike executing a statement while a ResultSet is open on the same connection. Low risk here (single row, within a transaction), but you may prefer to return a sentinel from the mapper and perform the expired-row delete after `queryForObject` returns. ########## components/camel-cassandraql/src/main/java/org/apache/camel/processor/keyvalue/cassandra/CassandraKeyValueRepository.java: ########## @@ -0,0 +1,369 @@ +/* + * 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.nio.ByteBuffer; +import java.time.Duration; +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.api.management.ManagedAttribute; +import org.apache.camel.api.management.ManagedOperation; +import org.apache.camel.api.management.ManagedResource; +import org.apache.camel.spi.Configurer; +import org.apache.camel.spi.KeyValueRepository; +import org.apache.camel.spi.Metadata; +import org.apache.camel.support.KeyValueRepositoryHelper; +import org.apache.camel.support.service.ServiceSupport; +import org.apache.camel.util.ObjectHelper; +import org.apache.camel.utils.cassandra.CassandraSessionHolder; +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]"); + } + + // ------------------------------------------------------------------------- + // 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 ? KeyValueRepositoryHelper.deserialize(buffer) : null; + } + + @Override + @ManagedOperation(description = "Put a key-value pair with optional TTL") + public Object put(String key, Object value, Duration ttl) { + LOGGER.debug("Putting key {} with TTL {}", key, ttl); + // Read the previous value before upserting + Object oldValue = get(key); + ByteBuffer serializedValue = KeyValueRepositoryHelper.serializeToByteBuffer(value); + int ttlSeconds = toTtlSeconds(ttl); + 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 ttl the time-to-live; {@code null}, zero, or negative means no expiration + * @return the existing value if the key was already present, or {@code null} if the put succeeded + */ + @Override + public Object putIfAbsent(String key, Object value, Duration ttl) { Review Comment: `putIfAbsent` nicely uses a lightweight transaction for atomicity, but `replace(key, expected, new, ttl)` and `delete(key, expected)` fall back to the non-atomic SPI default even though Cassandra could express them atomically via `UPDATE ... IF value = ?` / `DELETE ... IF value = ?`. Documented as "No" in the manual so not a defect — just flagging as a natural follow-up enhancement for true CAS on this backend. ########## components/camel-jpa/src/main/java/org/apache/camel/processor/keyvalue/jpa/JpaKeyValueRepository.java: ########## @@ -0,0 +1,511 @@ +/* + * 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.time.Duration; +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.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.KeyValueRepositoryHelper; +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] = KeyValueRepositoryHelper.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, Duration ttl) { + final Object[] rc = new Object[1]; + final EntityManager entityManager + = getTargetEntityManager(null, entityManagerFactory, false, sharedEntityManager, true); + + transactionStrategy.executeInTransaction(() -> { + if (joinTransaction) { + entityManager.joinTransaction(); + } + try { + long expiresAt = toExpiresAt(ttl); + byte[] serializedValue = KeyValueRepositoryHelper.serialize(value); + + KeyValueEntry entry = findByKey(entityManager, key); + if (entry != null) { + if (!entry.isExpired()) { + rc[0] = KeyValueRepositoryHelper.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] = KeyValueRepositoryHelper.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, Duration ttl) { + final Object[] rc = new Object[1]; + final EntityManager entityManager + = getTargetEntityManager(null, entityManagerFactory, false, sharedEntityManager, true); + + transactionStrategy.executeInTransaction(() -> { + if (joinTransaction) { + entityManager.joinTransaction(); + } + try { + long expiresAt = toExpiresAt(ttl); + byte[] serializedValue = KeyValueRepositoryHelper.serialize(value); + + KeyValueEntry entry = findByKey(entityManager, key); + if (entry != null && !entry.isExpired()) { + // key exists and is valid -- return existing value + rc[0] = KeyValueRepositoryHelper.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 ? KeyValueRepositoryHelper.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 long toExpiresAt(Duration ttl) { + if (ttl == null || ttl.isZero() || ttl.isNegative()) { + return 0; + } + return System.currentTimeMillis() + ttl.toMillis(); + } + + 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) { Review Comment: Minor style nit: `java.sql.SQLIntegrityConstraintViolationException` and `jakarta.persistence.EntityExistsException` are used as inline FQCNs here. The project Import Style rule prefers an `import` + simple name. Cosmetic only (CI is green, so OpenRewrite tolerated it). -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
