gnodet commented on code in PR #25993: URL: https://github.com/apache/camel/pull/25993#discussion_r3914269566
########## 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: _Claude Code on behalf of gnodet_ Good catch. Added try/catch rollback to both `delete()` and `clear()`, mirroring the pattern already used in `put()` and `putIfAbsent()`. For `clear()`, we snapshot the cache before clearing and restore it on broadcast failure. -- 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]
