gnodet commented on code in PR #25993: URL: https://github.com/apache/camel/pull/25993#discussion_r3914271249
########## 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: _Claude Code on behalf of gnodet_ Good point. Added a comment above the `LRUCacheFactory.newLRUCache()` call explaining the trade-off: when live keys exceed `maxCacheSize`, LRU-evicted entries are reported as absent by `get()`/`contains()` even though the value persists in the compacted topic. This matches the `KafkaIdempotentRepository` behavior and the comment advises users to increase `maxCacheSize` if they need a larger working set. -- 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]
