This is an automated email from the ASF dual-hosted git repository.
Croway pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/camel.git
The following commit(s) were added to refs/heads/main by this push:
new 155e967efb82 chore: camel-kafka - share the changelog topic handling
of the Kafka repositories
155e967efb82 is described below
commit 155e967efb82717b3ae89004cd6daae33f50de00
Author: croway <[email protected]>
AuthorDate: Wed Oct 7 15:02:36 2026 +0200
chore: camel-kafka - share the changelog topic handling of the Kafka
repositories
KafkaIdempotentRepository and KafkaKeyValueRepository had copies of the
same Kafka code: the consumer and producer setup, the read of the topic
on start, the background sync thread, the synchronous send and the stop.
KafkaChangelog now holds it, and each repository keeps its options, its
local cache, its record format and its rollback on a failed send.
KafkaChangelog is internal to the Kafka repositories.
The getters and setters of the options stay in each repository, as the
bean metadata is generated from the fields declared by the bean class.
The generated metadata is unchanged.
Co-Authored-By: Claude Opus 5.5 <[email protected]>
---
.../processor/idempotent/kafka/KafkaChangelog.java | 273 +++++++++++++++++++++
.../kafka/KafkaIdempotentRepository.java | 161 +-----------
.../keyvalue/kafka/KafkaKeyValueRepository.java | 162 ++----------
.../kafka/KafkaIdempotentRepositoryTest.java | 15 +-
4 files changed, 313 insertions(+), 298 deletions(-)
diff --git
a/components/camel-kafka/src/main/java/org/apache/camel/processor/idempotent/kafka/KafkaChangelog.java
b/components/camel-kafka/src/main/java/org/apache/camel/processor/idempotent/kafka/KafkaChangelog.java
new file mode 100644
index 000000000000..18940390ca95
--- /dev/null
+++
b/components/camel-kafka/src/main/java/org/apache/camel/processor/idempotent/kafka/KafkaChangelog.java
@@ -0,0 +1,273 @@
+/*
+ * 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.idempotent.kafka;
+
+import java.time.Duration;
+import java.util.Collection;
+import java.util.List;
+import java.util.Map;
+import java.util.Properties;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.function.Consumer;
+
+import org.apache.camel.CamelContext;
+import org.apache.camel.RuntimeCamelException;
+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.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.Deserializer;
+import org.apache.kafka.common.serialization.Serializer;
+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 used as the changelog of the local cache of a repository,
shared by the Kafka repositories
+ * ({@link KafkaIdempotentRepository} and the KafkaKeyValueRepository).
+ * <p/>
+ * On start, the whole topic is read from the beginning and every record is
applied to the cache; then, unless the
+ * repository syncs on startup only, the records written by the other
instances keep being applied by a background
+ * thread. The changes of this instance are written to the topic synchronously.
+ * <p/>
+ * This class is internal to the Kafka repositories of camel-kafka, and is not
meant to be used by applications.
+ *
+ * @param <V> the type of the record values
+ */
+public final class KafkaChangelog<V> {
+
+ private static final Logger LOG =
LoggerFactory.getLogger(KafkaChangelog.class);
+
+ private final String name;
+ private final String topic;
+ private final Properties consumerConfig;
+ private final Properties producerConfig;
+ private final Class<? extends Deserializer<V>> valueDeserializer;
+ private final Class<? extends Serializer<V>> valueSerializer;
+ private final int pollDurationMs;
+ private final boolean startupOnly;
+ private final Consumer<ConsumerRecord<String, V>> recordHandler;
+
+ private CamelContext camelContext;
+ private org.apache.kafka.clients.consumer.Consumer<String, V> consumer;
+ private Producer<String, V> producer;
+ private TopicPoller poller;
+ private ExecutorService executorService;
+
+ /**
+ * @param name the name of the repository, used in the logs
and in the name of the background thread
+ * @param topic the changelog topic
+ * @param consumerConfig the properties of the consumer, see {@link
#consumerConfig(Properties, String, String)}
+ * @param producerConfig the properties of the producer, see {@link
#producerConfig(Properties, String)}
+ * @param valueDeserializer the deserializer of the record values (the
keys are strings)
+ * @param valueSerializer the serializer of the record values (the keys
are strings)
+ * @param pollDurationMs the poll duration of the consumer
+ * @param startupOnly whether to read the topic on start only, or to
keep reading it in the background
+ * @param recordHandler applies a record of the topic to the local
cache
+ */
+ public KafkaChangelog(String name, String topic, Properties
consumerConfig, Properties producerConfig,
+ Class<? extends Deserializer<V>> valueDeserializer,
Class<? extends Serializer<V>> valueSerializer,
+ int pollDurationMs, boolean startupOnly,
Consumer<ConsumerRecord<String, V>> recordHandler) {
+ this.name = name;
+ this.topic = topic;
+ this.consumerConfig = consumerConfig;
+ this.producerConfig = producerConfig;
+ this.valueDeserializer = valueDeserializer;
+ this.valueSerializer = valueSerializer;
+ this.pollDurationMs = pollDurationMs;
+ this.startupOnly = startupOnly;
+ this.recordHandler = recordHandler;
+ }
+
+ /**
+ * The consumer properties of a repository: the configured properties,
otherwise properties with the bootstrap
+ * servers and the group id.
+ */
+ public static Properties consumerConfig(Properties consumerConfig, String
bootstrapServers, String groupId) {
+ if (consumerConfig != null) {
+ return consumerConfig;
+ }
+ Properties answer = new Properties();
+ StringHelper.notEmpty(bootstrapServers, "bootstrapServers");
+ answer.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
+ if (groupId != null) {
+ answer.put(ConsumerConfig.GROUP_ID_CONFIG, groupId);
+ }
+ return answer;
+ }
+
+ /**
+ * The producer properties of a repository: the configured properties,
otherwise properties with the bootstrap
+ * servers.
+ */
+ public static Properties producerConfig(Properties producerConfig, String
bootstrapServers) {
+ if (producerConfig != null) {
+ return producerConfig;
+ }
+ Properties answer = new Properties();
+ StringHelper.notEmpty(bootstrapServers, "bootstrapServers");
+ answer.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
+ return answer;
+ }
+
+ /**
+ * Creates the consumer and the producer, reads the topic from the
beginning, and starts the background thread
+ * unless the repository syncs on startup only.
+ *
+ * @param camelContext the CamelContext
+ * @param source the owner of the background thread
+ */
+ public void start(CamelContext camelContext, Object source) {
+ this.camelContext = camelContext;
+
+ consumerConfig.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG,
Boolean.FALSE.toString());
+ consumerConfig.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG,
StringDeserializer.class.getName());
+ consumerConfig.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG,
valueDeserializer.getName());
+ consumer = new KafkaConsumer<>(consumerConfig);
+
+ producerConfig.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG,
StringSerializer.class.getName());
+ producerConfig.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,
valueSerializer.getName());
+ // set up the producer to remove all batching on send, we want all
sends to be fully synchronous
+ producerConfig.putIfAbsent(ProducerConfig.ACKS_CONFIG, "1");
+ producerConfig.putIfAbsent(ProducerConfig.BATCH_SIZE_CONFIG, "0");
+ producer = new KafkaProducer<>(producerConfig);
+
+ poller = new TopicPoller();
+ ServiceHelper.startService(poller);
+ // populate cache on startup to be ready
+ StopWatch watch = new StopWatch();
+ LOG.info("Syncing {} from topic: {} starting", name, topic);
+ poller.run();
+ LOG.info("Syncing {} from topic: {} complete: {}", name, topic,
TimeUtils.printDuration(watch.taken(), true));
+
+ if (!startupOnly) {
+ // continue sync job in background
+ executorService =
camelContext.getExecutorServiceManager().newSingleThreadExecutor(source, name +
"Sync");
+ LOG.info("Syncing {} from topic: {} continuously using background
thread", name, topic);
+ executorService.submit(poller);
+ }
+ }
+
+ /**
+ * Stops the background thread, and closes the consumer and the producer.
+ */
+ public void stop() {
+ ServiceHelper.stopService(poller);
+ if (consumer != null) {
+ consumer.wakeup();
+ }
+ if (executorService != null && camelContext != null) {
+
camelContext.getExecutorServiceManager().shutdownNow(executorService);
+ executorService = null;
+ }
+ IOHelper.close(consumer, "consumer", LOG);
+ IOHelper.close(producer, "producer", LOG);
+ }
+
+ /**
+ * Writes a record to the topic, and waits until it is written.
+ */
+ public void send(String key, V value) {
+ try {
+ ObjectHelper.notNull(producer, "producer");
+ producer.send(new ProducerRecord<>(topic, key, value)).get(); //
sync send
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ throw new RuntimeCamelException(e);
+ } catch (ExecutionException e) {
+ throw new RuntimeCamelException(e);
+ }
+ }
+
+ /**
+ * To use the given producer, for testing.
+ */
+ void setProducer(Producer<String, V> producer) {
+ this.producer = producer;
+ }
+
+ 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, V> consumerRecords =
consumer.poll(Duration.ofMillis(pollDurationMs));
+ for (ConsumerRecord<String, V> consumerRecord : consumerRecords) {
+ recordHandler.accept(consumerRecord);
+ }
+ }
+ }
+
+ private class TopicPoller extends ServiceSupport implements Runnable {
+
+ private final AtomicBoolean init = new AtomicBoolean();
+
+ @Override
+ public void run() {
+ if (init.compareAndSet(false, true)) {
+ // sync cache on startup
+ LOG.debug("TopicPoller populating cache on startup");
+ populateCache();
+ LOG.debug("TopicPoller populated cache on startup complete");
+ return;
+ }
+
+ LOG.debug("TopicPoller running");
+ while (isRunAllowed()) {
+ try {
+ ConsumerRecords<String, V> consumerRecords =
consumer.poll(Duration.ofMillis(pollDurationMs));
+ for (ConsumerRecord<String, V> consumerRecord :
consumerRecords) {
+ recordHandler.accept(consumerRecord);
+ }
+ } catch (WakeupException e) {
+ LOG.debug("TopicPoller woken up during shutdown");
+ } catch (Exception e) {
+ LOG.warn("TopicPoller error syncing due to: {}. This
exception is ignored.", e.getMessage(), e);
+ }
+ }
+ LOG.debug("TopicPoller stopping");
+ }
+ }
+}
diff --git
a/components/camel-kafka/src/main/java/org/apache/camel/processor/idempotent/kafka/KafkaIdempotentRepository.java
b/components/camel-kafka/src/main/java/org/apache/camel/processor/idempotent/kafka/KafkaIdempotentRepository.java
index 4cd733efb073..438b94245c6e 100644
---
a/components/camel-kafka/src/main/java/org/apache/camel/processor/idempotent/kafka/KafkaIdempotentRepository.java
+++
b/components/camel-kafka/src/main/java/org/apache/camel/processor/idempotent/kafka/KafkaIdempotentRepository.java
@@ -16,44 +16,22 @@
*/
package org.apache.camel.processor.idempotent.kafka;
-import java.time.Duration;
-import java.util.Collection;
-import java.util.List;
import java.util.Map;
import java.util.Properties;
-import java.util.concurrent.ExecutionException;
-import java.util.concurrent.ExecutorService;
-import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicLong;
import org.apache.camel.CamelContext;
import org.apache.camel.CamelContextAware;
-import org.apache.camel.RuntimeCamelException;
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.IdempotentRepository;
import org.apache.camel.spi.Metadata;
import org.apache.camel.support.LRUCacheFactory;
-import org.apache.camel.support.service.ServiceHelper;
import org.apache.camel.support.service.ServiceSupport;
-import org.apache.camel.util.IOHelper;
import org.apache.camel.util.ObjectHelper;
-import org.apache.camel.util.StopWatch;
import org.apache.camel.util.StringHelper;
-import org.apache.camel.util.TimeUtils;
-import org.apache.kafka.clients.consumer.Consumer;
-import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.ConsumerRecord;
-import org.apache.kafka.clients.consumer.ConsumerRecords;
-import org.apache.kafka.clients.consumer.KafkaConsumer;
-import org.apache.kafka.clients.producer.KafkaProducer;
-import org.apache.kafka.clients.producer.Producer;
-import org.apache.kafka.clients.producer.ProducerConfig;
-import org.apache.kafka.clients.producer.ProducerRecord;
-import org.apache.kafka.common.PartitionInfo;
-import org.apache.kafka.common.TopicPartition;
-import org.apache.kafka.common.errors.WakeupException;
import org.apache.kafka.common.serialization.StringDeserializer;
import org.apache.kafka.common.serialization.StringSerializer;
import org.slf4j.Logger;
@@ -86,13 +64,10 @@ public class KafkaIdempotentRepository extends
ServiceSupport implements Idempot
private static final int DEFAULT_POLL_DURATION_MS = 100;
private CamelContext camelContext;
- private ExecutorService executorService;
- private TopicPoller poller;
private final AtomicLong cacheCounter = new AtomicLong();
// internal properties
private Map<String, Object> cache;
- private Consumer<String, String> consumer;
- private Producer<String, String> producer;
+ private KafkaChangelog<String> changelog;
private Properties producerConfig;
private Properties consumerConfig;
@@ -331,125 +306,23 @@ public class KafkaIdempotentRepository extends
ServiceSupport implements Idempot
this.cache = LRUCacheFactory.newLRUCache(maxCacheSize);
- if (consumerConfig == null) {
- consumerConfig = new Properties();
- StringHelper.notEmpty(bootstrapServers, "bootstrapServers");
- consumerConfig.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG,
bootstrapServers);
- if (groupId != null) {
- consumerConfig.put(ConsumerConfig.GROUP_ID_CONFIG, groupId);
- }
- }
-
- if (producerConfig == null) {
- producerConfig = new Properties();
- StringHelper.notEmpty(bootstrapServers, "bootstrapServers");
- producerConfig.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG,
bootstrapServers);
- }
-
- ObjectHelper.notNull(consumerConfig, "consumerConfig");
- ObjectHelper.notNull(producerConfig, "producerConfig");
-
- consumerConfig.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG,
Boolean.FALSE.toString());
- consumerConfig.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG,
StringDeserializer.class.getName());
- consumerConfig.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG,
StringDeserializer.class.getName());
-
- consumer = new KafkaConsumer<>(consumerConfig);
-
- producerConfig.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG,
StringSerializer.class.getName());
- producerConfig.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,
StringSerializer.class.getName());
- // set up the producer to remove all batching on send, we want all
sends
- // to be fully synchronous
- producerConfig.putIfAbsent(ProducerConfig.ACKS_CONFIG, "1");
- producerConfig.putIfAbsent(ProducerConfig.BATCH_SIZE_CONFIG, "0");
- producer = new KafkaProducer<>(producerConfig);
-
- poller = new TopicPoller();
- ServiceHelper.startService(poller);
- // populate cache on startup to be ready
- StopWatch watch = new StopWatch();
- LOG.info("Syncing KafkaIdempotentRepository from topic: {} starting",
topic);
- poller.run();
- LOG.info("Syncing KafkaIdempotentRepository from topic: {} complete:
{}", topic,
- TimeUtils.printDuration(watch.taken(), true));
-
- if (!startupOnly) {
- // continue sync job in background
- executorService
- =
camelContext.getExecutorServiceManager().newSingleThreadExecutor(this,
"KafkaIdempotentRepositorySync");
- LOG.info("Syncing KafkaIdempotentRepository from topic: {}
continuously using background thread", topic);
- executorService.submit(poller);
- }
+ consumerConfig = KafkaChangelog.consumerConfig(consumerConfig,
bootstrapServers, groupId);
+ producerConfig = KafkaChangelog.producerConfig(producerConfig,
bootstrapServers);
+ changelog = new KafkaChangelog<>(
+ "KafkaIdempotentRepository", topic, consumerConfig,
producerConfig, StringDeserializer.class,
+ StringSerializer.class, pollDurationMs,
+ startupOnly, this::addToCache);
+ changelog.start(camelContext, this);
}
@Override
protected void doStop() throws Exception {
- ServiceHelper.stopService(poller);
- if (consumer != null) {
- consumer.wakeup();
+ if (changelog != null) {
+ changelog.stop();
}
- if (executorService != null && camelContext != null) {
-
camelContext.getExecutorServiceManager().shutdownNow(executorService);
- executorService = null;
- }
- IOHelper.close(consumer, "consumer", LOG);
- IOHelper.close(producer, "producer", LOG);
LOG.debug("Stopped KafkaIdempotentRepository. Cache counter: {}",
cacheCounter.get());
}
- 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, String> consumerRecords =
consumer.poll(Duration.ofMillis(pollDurationMs));
- for (ConsumerRecord<String, String> consumerRecord :
consumerRecords) {
- addToCache(consumerRecord);
- }
- }
- }
-
- private class TopicPoller extends ServiceSupport implements Runnable {
-
- private final AtomicBoolean init = new AtomicBoolean();
-
- @Override
- public void run() {
- if (init.compareAndSet(false, true)) {
- // sync cache on startup
- LOG.debug("TopicPoller populating cache on startup");
- populateCache();
- LOG.debug("TopicPoller populated cache on startup complete");
- return;
- }
-
- LOG.debug("TopicPoller running");
- while (isRunAllowed()) {
- try {
- ConsumerRecords<String, String> consumerRecords =
consumer.poll(Duration.ofMillis(pollDurationMs));
- for (ConsumerRecord<String, String> consumerRecord :
consumerRecords) {
- addToCache(consumerRecord);
- }
- } catch (WakeupException e) {
- LOG.debug("TopicPoller woken up during shutdown");
- } catch (Exception e) {
- LOG.warn("TopicPoller error syncing due to: " +
e.getMessage() + ". This exception is ignored.", e);
- }
- }
- LOG.debug("TopicPoller stopping");
- }
- }
-
private void addToCache(ConsumerRecord<String, String> consumerRecord) {
cacheCounter.incrementAndGet();
CacheAction action;
@@ -490,17 +363,9 @@ public class KafkaIdempotentRepository extends
ServiceSupport implements Idempot
}
private void broadcastAction(String key, CacheAction action) {
- try {
- LOG.debug("Broadcasting action:{} for key:{}", action, key);
- ObjectHelper.notNull(producer, "producer");
-
- producer.send(new ProducerRecord<>(topic, key,
action.toString())).get(); // sync send
- } catch (InterruptedException e) {
- Thread.currentThread().interrupt();
- throw new RuntimeCamelException(e);
- } catch (ExecutionException e) {
- throw new RuntimeCamelException(e);
- }
+ LOG.debug("Broadcasting action:{} for key:{}", action, key);
+ ObjectHelper.notNull(changelog, "changelog");
+ changelog.send(key, action.toString());
}
@Override
diff --git
a/components/camel-kafka/src/main/java/org/apache/camel/processor/keyvalue/kafka/KafkaKeyValueRepository.java
b/components/camel-kafka/src/main/java/org/apache/camel/processor/keyvalue/kafka/KafkaKeyValueRepository.java
index ada853fa5503..707b55f87032 100644
---
a/components/camel-kafka/src/main/java/org/apache/camel/processor/keyvalue/kafka/KafkaKeyValueRepository.java
+++
b/components/camel-kafka/src/main/java/org/apache/camel/processor/keyvalue/kafka/KafkaKeyValueRepository.java
@@ -20,15 +20,10 @@ 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;
@@ -38,35 +33,18 @@ 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.processor.idempotent.kafka.KafkaChangelog;
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;
@@ -111,14 +89,11 @@ public class KafkaKeyValueRepository extends
ServiceSupport implements KeyValueR
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;
+ private KafkaChangelog<byte[]> changelog;
@Metadata(description = "Custom properties for the Kafka consumer")
private Properties consumerConfig;
@@ -326,16 +301,9 @@ public class KafkaKeyValueRepository extends
ServiceSupport implements KeyValueR
}
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);
- }
+ LOG.debug("Broadcasting to topic {} for key {}", topic, key);
+ ObjectHelper.notNull(changelog, "changelog");
+ changelog.send(key, payload);
}
//
-------------------------------------------------------------------------
@@ -405,29 +373,6 @@ public class KafkaKeyValueRepository extends
ServiceSupport implements KeyValueR
}
}
- 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;
@@ -461,102 +406,23 @@ public class KafkaKeyValueRepository extends
ServiceSupport implements KeyValueR
// if you need a larger working set to be visible locally.
this.cache = LRUCacheFactory.newLRUCache(maxCacheSize);
- if (consumerConfig == null) {
- consumerConfig = new Properties();
- StringHelper.notEmpty(bootstrapServers, "bootstrapServers");
- consumerConfig.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG,
bootstrapServers);
- if (groupId != null) {
- consumerConfig.put(ConsumerConfig.GROUP_ID_CONFIG, groupId);
- }
- }
-
- if (producerConfig == null) {
- producerConfig = new Properties();
- StringHelper.notEmpty(bootstrapServers, "bootstrapServers");
- producerConfig.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG,
bootstrapServers);
- }
-
- ObjectHelper.notNull(consumerConfig, "consumerConfig");
- ObjectHelper.notNull(producerConfig, "producerConfig");
-
- consumerConfig.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG,
Boolean.FALSE.toString());
- consumerConfig.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG,
StringDeserializer.class.getName());
- consumerConfig.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG,
ByteArrayDeserializer.class.getName());
-
- consumer = new KafkaConsumer<>(consumerConfig);
-
- producerConfig.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG,
StringSerializer.class.getName());
- producerConfig.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,
ByteArraySerializer.class.getName());
- producerConfig.putIfAbsent(ProducerConfig.ACKS_CONFIG, "1");
- producerConfig.putIfAbsent(ProducerConfig.BATCH_SIZE_CONFIG, "0");
- producer = new KafkaProducer<>(producerConfig);
-
- poller = new TopicPoller();
- ServiceHelper.startService(poller);
- // populate cache on startup
- StopWatch watch = new StopWatch();
- LOG.info("Syncing KafkaKeyValueRepository from topic: {} starting",
topic);
- poller.run();
- LOG.info("Syncing KafkaKeyValueRepository from topic: {} complete:
{}", topic,
- TimeUtils.printDuration(watch.taken(), true));
-
- if (!startupOnly) {
- executorService
- =
camelContext.getExecutorServiceManager().newSingleThreadExecutor(this,
"KafkaKeyValueRepositorySync");
- LOG.info("Syncing KafkaKeyValueRepository from topic: {}
continuously using background thread", topic);
- executorService.submit(poller);
- }
+ consumerConfig = KafkaChangelog.consumerConfig(consumerConfig,
bootstrapServers, groupId);
+ producerConfig = KafkaChangelog.producerConfig(producerConfig,
bootstrapServers);
+ changelog = new KafkaChangelog<>(
+ "KafkaKeyValueRepository", topic, consumerConfig,
producerConfig, ByteArrayDeserializer.class,
+ ByteArraySerializer.class, pollDurationMs,
+ startupOnly, this::addToCache);
+ changelog.start(camelContext, this);
}
@Override
protected void doStop() throws Exception {
- ServiceHelper.stopService(poller);
- if (consumer != null) {
- consumer.wakeup();
- }
- if (executorService != null && camelContext != null) {
-
camelContext.getExecutorServiceManager().shutdownNow(executorService);
- executorService = null;
+ if (changelog != null) {
+ changelog.stop();
}
- IOHelper.close(consumer, "consumer", LOG);
- IOHelper.close(producer, "producer", LOG);
LOG.debug("Stopped KafkaKeyValueRepository. Cache counter: {}",
cacheCounter.get());
}
- //
-------------------------------------------------------------------------
- // TopicPoller inner class
- //
-------------------------------------------------------------------------
-
- private class TopicPoller extends ServiceSupport implements Runnable {
-
- private final AtomicBoolean init = new AtomicBoolean();
-
- @Override
- public void run() {
- if (init.compareAndSet(false, true)) {
- LOG.debug("TopicPoller populating cache on startup");
- populateCache();
- LOG.debug("TopicPoller populated cache on startup complete");
- return;
- }
-
- LOG.debug("TopicPoller running");
- while (isRunAllowed()) {
- try {
- ConsumerRecords<String, byte[]> consumerRecords =
consumer.poll(Duration.ofMillis(pollDurationMs));
- for (ConsumerRecord<String, byte[]> consumerRecord :
consumerRecords) {
- addToCache(consumerRecord);
- }
- } catch (WakeupException e) {
- LOG.debug("TopicPoller woken up during shutdown");
- } catch (Exception e) {
- LOG.warn("TopicPoller error syncing due to: " +
e.getMessage() + ". This exception is ignored.", e);
- }
- }
- LOG.debug("TopicPoller stopping");
- }
- }
-
//
-------------------------------------------------------------------------
// CacheEntry
//
-------------------------------------------------------------------------
diff --git
a/components/camel-kafka/src/test/java/org/apache/camel/processor/idempotent/kafka/KafkaIdempotentRepositoryTest.java
b/components/camel-kafka/src/test/java/org/apache/camel/processor/idempotent/kafka/KafkaIdempotentRepositoryTest.java
index e3badc9bcdfb..d40db9e53974 100644
---
a/components/camel-kafka/src/test/java/org/apache/camel/processor/idempotent/kafka/KafkaIdempotentRepositoryTest.java
+++
b/components/camel-kafka/src/test/java/org/apache/camel/processor/idempotent/kafka/KafkaIdempotentRepositoryTest.java
@@ -18,12 +18,14 @@ package org.apache.camel.processor.idempotent.kafka;
import java.lang.reflect.Field;
import java.util.Map;
+import java.util.Properties;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.TimeUnit;
import org.apache.camel.RuntimeCamelException;
import org.apache.kafka.clients.producer.MockProducer;
import org.apache.kafka.clients.producer.ProducerRecord;
+import org.apache.kafka.common.serialization.StringDeserializer;
import org.apache.kafka.common.serialization.StringSerializer;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
@@ -49,7 +51,7 @@ public class KafkaIdempotentRepositoryTest {
repository.setTopic("test-topic");
setField(repository, "cache", cache);
- setField(repository, "producer", mockProducer);
+ setField(repository, "changelog", changelog(mockProducer));
}
@Test
@@ -79,7 +81,7 @@ public class KafkaIdempotentRepositoryTest {
void testAddRollsBackCacheOnBroadcastFailure() throws Exception {
MockProducer<String, String> failingProducer
= new MockProducer<>(false, null, new StringSerializer(), new
StringSerializer());
- setField(repository, "producer", failingProducer);
+ setField(repository, "changelog", changelog(failingProducer));
Thread sender = new Thread(() -> {
try {
@@ -142,6 +144,15 @@ public class KafkaIdempotentRepositoryTest {
assertTrue(repository.add("key1"));
}
+ private static KafkaChangelog<String> changelog(MockProducer<String,
String> producer) {
+ KafkaChangelog<String> changelog = new KafkaChangelog<>(
+ "KafkaIdempotentRepository", "test-topic", new Properties(),
new Properties(), StringDeserializer.class,
+ StringSerializer.class, 100, true, record -> {
+ });
+ changelog.setProducer(producer);
+ return changelog;
+ }
+
private static void setField(Object target, String fieldName, Object
value) throws Exception {
Field field = target.getClass().getDeclaredField(fieldName);
field.setAccessible(true);
