You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@camel.apache.org by "Croway (via GitHub)" <gi...@apache.org> on 2026/10/08 07:02:53 UTC
[PR] camel-kafka - share the changelog topic handling of the Kafka repositories [camel]
Croway opened a new pull request, #27543:
URL: https://github.com/apache/camel/pull/27543
## Description
Refactor only. `KafkaIdempotentRepository` and `KafkaKeyValueRepository` had copies of the same Kafka code (CPD flagged about 260 duplicated lines across 7 blocks):
- the consumer and producer properties (bootstrap servers and group id defaults, `enable.auto.commit=false`, `acks`/`batch.size` defaults, key and value serdes);
- reading the topic from the beginning on start, and the background sync thread (`TopicPoller`);
- the synchronous send to the topic;
- stopping (wakeup, executor shutdown, closing the clients).
That code is now in `KafkaChangelog<V>`, next to `KafkaConsumerUtil` in `org.apache.camel.processor.idempotent.kafka`, with `V` being the record value type (`String` for the idempotent repository, `byte[]` for the key-value one). Each repository keeps its options, its local cache, its record format and its rollback when a send fails. The log messages and the name of the background thread are unchanged.
The getters and setters of the options stay in each repository: the bean metadata generator (`GeneratePojoBeanMojo`) only reads the fields declared on the bean class, so moving the options to a shared class would remove them from the bean documentation. The generated bean metadata and configurers are unchanged.
`KafkaIdempotentRepositoryTest` set the private `producer` field by reflection; it now injects a `KafkaChangelog` built with its `MockProducer`.
This was split out of #27495 (Kafka share groups, CAMEL-25029) at review, since it is unrelated to share groups.
### Verification
- `camel-kafka` unit tests: 220, green. `-Psourcecheck` green.
- Repository ITs: `KafkaIdempotentRepositoryEagerIT`, `KafkaIdempotentRepositoryNonEagerIT`, `KafkaIdempotentRepositoryPersistenceIT`, `KafkaConsumerIdempotent*IT` (4) and `KafkaKeyValueRepositoryIT` (23 tests), all green.
## Target
- [x] I checked that the commit is targeting the correct branch (Camel 4 uses the `main` branch)
## Tracking
- [ ] If this is a large change, bug fix, or code improvement, I checked there is a [JIRA issue](https://issues.apache.org/jira/browse/CAMEL) filed for the change (usually before you start working on it). (Internal refactor, no behaviour change.)
## Apache Camel coding standards and style
- [x] I checked that each commit in the pull request has a meaningful subject line and body.
- [x] I have run `mvn clean install -DskipTests` locally from root folder and I have committed all auto-generated changes.
_Claude Code on behalf of Croway_
--
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: commits-unsubscribe@camel.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] camel-kafka - share the changelog topic handling of the Kafka repositories [camel]
Posted by "gnodet-bot (via GitHub)" <gi...@apache.org>.
gnodet-bot commented on code in PR #27543:
URL: https://github.com/apache/camel/pull/27543#discussion_r4216076544
##########
components/camel-kafka/src/main/java/org/apache/camel/processor/idempotent/kafka/KafkaChangelog.java:
##########
@@ -0,0 +1,271 @@
+/*
+ * 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.
+ *
+ * @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: " + e.getMessage() + ". This exception is ignored.", e);
Review Comment:
⚠️ **Eager string concatenation in `LOG.warn`:** The message is built via `+` even when `WARN` is disabled. Prefer the parameterized form so the concatenation is deferred:
```suggestion
LOG.warn("TopicPoller error syncing due to: {}. This exception is ignored.", e.getMessage(), e);
```
##########
components/camel-kafka/src/main/java/org/apache/camel/processor/idempotent/kafka/KafkaChangelog.java:
##########
@@ -0,0 +1,271 @@
+/*
+ * 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.
+ *
+ * @param <V> the type of the record values
+ */
+public final class KafkaChangelog<V> {
+
Review Comment:
💡 **Package placement:** `KafkaChangelog` is shared infrastructure — it's imported by `KafkaKeyValueRepository` which lives in `processor.keyvalue.kafka`, yet the class itself lives under `processor.idempotent.kafka`. This creates an awkward cross-package dependency on the idempotent package for something that isn't idempotent-specific. Consider placing it in a neutral parent package (e.g. `processor.kafka.changelog` or `processor.kafka.common`) so neither consumer has to reach into the other's namespace.
--
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: commits-unsubscribe@camel.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] camel-kafka - share the changelog topic handling of the Kafka repositories [camel]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #27543:
URL: https://github.com/apache/camel/pull/27543#issuecomment-6055116818
<!-- ci-tested-modules -->
:test_tube: **CI tested the following changed modules:**
- `components/camel-kafka`
---
<details><summary>:microscope: Scalpel shadow comparison — Scalpel: 9 of 700 tested, 27 compile-only — current: 9 all tested</summary>
[Maveniverse Scalpel](https://github.com/maveniverse/scalpel) detected **9 affected modules** (current approach: 9).
**Skip-tests mode would test 9 modules** (1 direct + 8 downstream), **skip tests for 27** (generated code, meta-modules)
<details><summary>Modules Scalpel would test (9)</summary>
- `camel-jbang-mcp` ← downstream of org.apache.camel:camel-catalog
- `camel-jbang-plugin-mcp` ← downstream of org.apache.camel:camel-jbang-core
- `camel-jbang-plugin-route-parser` ← downstream of org.apache.camel:camel-route-parser
- `camel-jbang-plugin-tui` ← downstream of org.apache.camel:camel-catalog
- `camel-jbang-plugin-validate` ← downstream of org.apache.camel:camel-yaml-dsl-validator
- `camel-kafka` ← components/camel-kafka/src/main/java/org/apache/camel/processor/idempotent/kafka/KafkaChangelog.java, components/camel-kafka/src/main/java/org/apache/camel/processor/idempotent/kafka/KafkaIdempotentRepository.java, components/camel-kafka/src/main/java/org/apache/camel/processor/keyvalue/kafka/KafkaKeyValueRepository.java, components/camel-kafka/src/test/java/org/apache/camel/processor/idempotent/kafka/KafkaIdempotentRepositoryTest.java
- `camel-launcher-container` ← downstream of org.apache.camel:camel-launcher
- `camel-yaml-dsl-validator` ← downstream of org.apache.camel:camel-catalog
- `camel-yaml-dsl-validator-maven-plugin` ← downstream of org.apache.camel:camel-yaml-dsl-validator
</details>
<details><summary>Modules with tests skipped (27)</summary>
- `apache-camel`
- `camel-allcomponents`
- `camel-catalog`
- `camel-catalog-console`
- `camel-catalog-maven`
- `camel-catalog-suggest`
- `camel-componentdsl`
- `camel-endpointdsl`
- `camel-endpointdsl-support`
- `camel-itest`
- `camel-jbang-core`
- `camel-jbang-it`
- `camel-jbang-main`
- `camel-jbang-plugin-edit`
- `camel-jbang-plugin-generate`
- `camel-jbang-plugin-kubernetes`
- `camel-jbang-plugin-test`
- `camel-kamelet-main`
- `camel-launcher`
- `camel-report-maven-plugin`
- `camel-route-parser`
- `camel-yaml-dsl`
- `camel-yaml-dsl-deserializers`
- `camel-yaml-dsl-maven-plugin`
- `coverage`
- `docs`
- `dummy-component`
</details>
> :information_source: Shadow mode — Scalpel observes but does not affect test execution. [Learn more](https://github.com/maveniverse/scalpel)
</details>
:warning: **Some tests are disabled on GitHub Actions** (`@DisabledIfSystemProperty(named = "ci.env.name")`) and require manual verification:
- `components/camel-kafka`: 2 test(s) disabled on GitHub Actions
<details><summary>All tested modules (36 modules, 5m 50s total)</summary>
**Total reactor time:** 5m 50s
| Module | Duration | Status |
| --- | --- | --- |
| Camel :: Launcher | 53.3s | SUCCESS |
| Camel :: JBang :: Plugin :: TUI | 45.4s | SUCCESS |
| Camel :: JBang :: MCP | 41.3s | SUCCESS |
| Camel :: Catalog :: Camel Catalog | 24.6s | SUCCESS |
| Camel :: Component DSL | 21.9s | SUCCESS |
| Camel :: YAML DSL | 18.9s | SUCCESS |
| Camel :: YAML DSL :: Validator | 18.9s | SUCCESS |
| Camel :: JBang :: Plugin :: Validate | 16.8s | SUCCESS |
| Camel :: Docs | 16.0s | SUCCESS |
| Camel :: JBang :: Plugin :: Testing | 14.5s | SUCCESS |
| Camel :: Kamelet Main | 11.7s | SUCCESS |
| Camel :: JBang :: Plugin :: Kubernetes | 10.4s | SUCCESS |
| Camel :: Catalog :: Camel Route Parser | 8.5s | SUCCESS |
| Camel :: Catalog :: Camel Report Maven Plugin | 8.2s | SUCCESS |
| Camel :: YAML DSL :: Deserializers | 7.6s | SUCCESS |
| Camel :: All Components Sync point | 6.2s | SUCCESS |
| Camel :: YAML DSL :: Validator Maven Plugin | 5.0s | SUCCESS |
| Camel :: YAML DSL :: Maven Plugins | 3.0s | SUCCESS |
| Camel :: Catalog :: Maven | 2.4s | SUCCESS |
| Camel :: Catalog :: Suggest (deprecated) | 2.3s | SUCCESS |
| Camel :: Catalog :: Dummy Component | 1.8s | SUCCESS |
| Camel :: Coverage | 1.6s | SUCCESS |
| Camel :: JBang :: Plugin :: Edit | 1.5s | SUCCESS |
| Camel :: JBang :: Plugin :: Generate | 1.2s | SUCCESS |
| Camel :: Assembly | 1.2s | SUCCESS |
| Camel :: JBang :: Plugin :: Route Parser | 1.0s | SUCCESS |
| Camel :: JBang :: Integration tests | 1.0s | SUCCESS |
| Camel :: JBang :: Main | 1.0s | SUCCESS |
| Camel :: Catalog :: Console | 0.9s | SUCCESS |
| Camel :: Endpoint DSL :: Support | 0.8s | SUCCESS |
| Camel :: Launcher :: Container | 0.8s | SUCCESS |
| Camel :: JBang :: Plugin :: MCP | 0.8s | SUCCESS |
| Camel :: Endpoint DSL | n/a | |
| Camel :: Integration Tests | n/a | |
| Camel :: JBang :: Core | n/a | |
| Camel :: Kafka | n/a | |
**Top 20 slowest modules:**
- `Camel :: Launcher` (53.3s)
- `Camel :: JBang :: Plugin :: TUI` (45.4s)
- `Camel :: JBang :: MCP` (41.3s)
- `Camel :: Catalog :: Camel Catalog` (24.6s)
- `Camel :: Component DSL` (21.9s)
- `Camel :: YAML DSL` (18.9s)
- `Camel :: YAML DSL :: Validator` (18.9s)
- `Camel :: JBang :: Plugin :: Validate` (16.8s)
- `Camel :: Docs` (16.0s)
- `Camel :: JBang :: Plugin :: Testing` (14.5s)
- `Camel :: Kamelet Main` (11.7s)
- `Camel :: JBang :: Plugin :: Kubernetes` (10.4s)
- `Camel :: Catalog :: Camel Route Parser` (8.5s)
- `Camel :: Catalog :: Camel Report Maven Plugin` (8.2s)
- `Camel :: YAML DSL :: Deserializers` (7.6s)
- `Camel :: All Components Sync point` (6.2s)
- `Camel :: YAML DSL :: Validator Maven Plugin` (5.0s)
- `Camel :: YAML DSL :: Maven Plugins` (3.0s)
- `Camel :: Catalog :: Maven` (2.4s)
- `Camel :: Catalog :: Suggest (deprecated)` (2.3s)
</details>
---
:gear: [View full build and test results](https://github.com/apache/camel/actions/runs/37741061607)
--
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: commits-unsubscribe@camel.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] chore: camel-kafka - share the changelog topic handling of the Kafka repositories [camel]
Posted by "Croway (via GitHub)" <gi...@apache.org>.
Croway merged PR #27543:
URL: https://github.com/apache/camel/pull/27543
--
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: commits-unsubscribe@camel.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] chore: camel-kafka - share the changelog topic handling of the Kafka repositories [camel]
Posted by "Croway (via GitHub)" <gi...@apache.org>.
Croway commented on PR #27543:
URL: https://github.com/apache/camel/pull/27543#issuecomment-6058786435
ci failures are unrelated (TUI), merging
--
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: commits-unsubscribe@camel.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] camel-kafka - share the changelog topic handling of the Kafka repositories [camel]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #27543:
URL: https://github.com/apache/camel/pull/27543#issuecomment-6054460275
:star2: Thank you for your contribution to the Apache Camel project! :star2:
:robot: CI automation will test this PR automatically.
:camel: Apache Camel Committers, please review the following items:
* First-time contributors **require MANUAL approval** for the GitHub Actions to run
* You can use the command `/component-test (camel-)component-name1 (camel-)component-name2..` to request a test from the test bot although they are normally detected and executed by CI.
* You can label PRs using `skip-tests` and `test-dependents` to fine-tune the checks executed by this PR.
* Build and test logs are available in the summary page. **Only** [Apache Camel committers](https://camel.apache.org/community/team/#committers) have access to the summary.
:warning: Be careful when sharing logs. Review their contents before sharing them publicly.
--
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: commits-unsubscribe@camel.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] camel-kafka - share the changelog topic handling of the Kafka repositories [camel]
Posted by "davsclaus (via GitHub)" <gi...@apache.org>.
davsclaus commented on code in PR #27543:
URL: https://github.com/apache/camel/pull/27543#discussion_r4216654009
##########
components/camel-kafka/src/main/java/org/apache/camel/processor/keyvalue/kafka/KafkaKeyValueRepository.java:
##########
@@ -461,102 +406,27 @@ protected void doStart() throws Exception {
// 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
Review Comment:
Nit: this section header is left without anything under it now that `TopicPoller` moved to `KafkaChangelog`, so the three comment lines (and the blank line after them) can go.
--
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: commits-unsubscribe@camel.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org