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/07 09:22:04 UTC

[PR] CAMEL-25029: camel-kafka - extract a reusable client configuration layer [camel]

Croway opened a new pull request, #27485:
URL: https://github.com/apache/camel/pull/27485

   ## Description
   
   [CAMEL-25029](https://issues.apache.org/jira/browse/CAMEL-25029): first of three PRs for the Kafka share group consumer (KIP-932). The plan is in [this JIRA comment](https://issues.apache.org/jira/browse/CAMEL-25029?focusedCommentId=18124365#comment-18124365):
   
   1. **This PR:** extract a reusable client configuration layer inside `camel-kafka`. Refactor only.
   2. Move the shared layer into a flat `camel-kafka-common` module.
   3. Add the `kafka-share` consumer, built on the common layer.
   
   The share consumer cannot reuse `KafkaConfiguration`. `ShareConsumerConfig` throws a `ConfigException` on `auto.offset.reset`, `enable.auto.commit`, `partition.assignment.strategy`, `session.timeout.ms`, `heartbeat.interval.ms` and `group.protocol`, and `createConsumerProperties()` always sets them. So the options that every Kafka client shares move to a base class.
   
   ### Changes
   
   - **`KafkaClientConfiguration`** (new, abstract, `@UriParams`) holds 54 shared options:
     - brokers and client id
     - connection, metrics and backoff
     - SSL, SASL, Kerberos and OAuth
     - deserializers, header filter and header deserializer
     - schema registry, shutdown timeout and additional properties
   
     It also holds the code that writes these options to the client properties: `applyCommonClientProperties`, `applySecurityProperties` and `applyAdditionalProperties`. The consumer and producer SSL builders were two copies of the same logic and are now one method. `KafkaConfiguration` extends the new class and keeps all its public getters and setters.
   - **`AbstractKafkaComponent`** (new) holds the client factory, the poll exception strategy, the `createConsumerBackoff*` options, global SSL and the deferred start of consumers. `KafkaComponent` extends it and keeps the configuration, the manual commit factory and the deprecated `subscribeConsumer*` options.
   - `KafkaSecurityConfigurer.configure(...)` and `KafkaRecordProcessor.propagateHeaders(...)` now take `KafkaClientConfiguration`. This is source compatible for callers that pass a `KafkaConfiguration`.
   - **Relabelled options:** `connectionMaxIdleMs`, `receiveBufferBytes`, `sendBufferBytes`, `metricReporters`, `noOfMetricsSample`, `metricsSampleWindowMs` and `reconnectBackoffMs` change from `producer` to `common`. They were always written to the consumer properties too. The catalog generator drops `producer` options from consumer-only endpoints, so `kafka-share` would lose them otherwise. This is the only change in `kafka.json`. In the endpoint DSL the options are added to the consumer builder and stay on the producer builder.
   
   `KafkaClientFactory.getBrokers(KafkaConfiguration)` is unchanged on purpose: camel-quarkus overrides it. PR 3 will add a separate overload for the share consumer.
   
   ### Verification
   
   - Before the refactor I captured the consumer and producer client properties for 8 configurations: defaults, many options, SSL options, `sslContextParameters`, SASL, and `saslAuthType` SCRAM, OAuth and NONE+SSL. After the refactor the output is identical. That snapshot test was temporary and is not part of this PR.
   - New `KafkaClientConfigurationTest` checks that:
     - the shared options reach both clients
     - the SSL, `sslContextParameters`, SASL and `saslAuthType` properties are the same for the producer and the consumer
     - additional properties override the shared options
     - `copy()` gives the copy its own additional properties
   - `mvn install` on `components/camel-kafka`: 232 unit tests and 128 integration tests (default profile), all green.
   
   No upgrade guide entry: nothing is removed or renamed, and the only user-visible change is the label of seven options.
   
   ## Target
   
   - [x] I checked that the commit is targeting the correct branch (Camel 4 uses the `main` branch)
   
   ## Tracking
   
   - [x] 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).
   
   ## 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-25029: camel-kafka - extract a reusable client configuration layer [camel]

Posted by "gnodet-bot (via GitHub)" <gi...@apache.org>.
gnodet-bot commented on code in PR #27485:
URL: https://github.com/apache/camel/pull/27485#discussion_r4205727890


##########
components/camel-kafka/src/test/java/org/apache/camel/component/kafka/KafkaClientConfigurationTest.java:
##########
@@ -0,0 +1,199 @@
+/*
+ * 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.component.kafka;
+
+import java.util.Map;
+import java.util.Properties;
+import java.util.TreeMap;
+
+import org.apache.camel.component.kafka.security.KafkaAuthType;
+import org.apache.camel.support.jsse.KeyManagersParameters;
+import org.apache.camel.support.jsse.KeyStoreParameters;
+import org.apache.camel.support.jsse.SSLContextParameters;
+import org.apache.camel.support.jsse.TrustManagersParameters;
+import org.apache.kafka.clients.CommonClientConfigs;
+import org.apache.kafka.common.config.SaslConfigs;
+import org.apache.kafka.common.config.SslConfigs;
+import org.junit.jupiter.api.Test;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertNotSame;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+/**
+ * The options of {@link KafkaClientConfiguration} must be written the same way to every Kafka client properties.
+ */
+class KafkaClientConfigurationTest {
+
+    @Test
+    void commonClientOptionsAreAppliedToProducerAndConsumer() {

Review Comment:
   ✅ Good test coverage — 6 tests covering common properties parity, SSL/SASL parity, `SSLContextParameters` parity, `saslAuthType` parity, additional properties override, and copy isolation. The `assertEquals(producer, consumer)` assertion for security properties (around line 96) is particularly valuable as it would catch any future drift between producer and consumer SSL configuration.



##########
components/camel-kafka/src/main/java/org/apache/camel/component/kafka/KafkaClientConfiguration.java:
##########
@@ -0,0 +1,1178 @@
+/*
+ * 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.component.kafka;
+
+import java.util.HashMap;
+import java.util.List;
+import java.util.Locale;
+import java.util.Map;
+import java.util.Objects;
+import java.util.Properties;
+import java.util.stream.Collectors;
+
+import org.apache.camel.RuntimeCamelException;
+import org.apache.camel.component.kafka.security.KafkaAuthType;
+import org.apache.camel.component.kafka.security.KafkaSecurityConfigurer;
+import org.apache.camel.component.kafka.serde.DefaultKafkaHeaderDeserializer;
+import org.apache.camel.component.kafka.serde.KafkaHeaderDeserializer;
+import org.apache.camel.spi.HeaderFilterStrategy;
+import org.apache.camel.spi.HeaderFilterStrategyAware;
+import org.apache.camel.spi.UriParam;
+import org.apache.camel.spi.UriParams;
+import org.apache.camel.support.ResourceHelper;
+import org.apache.camel.support.jsse.CipherSuitesParameters;
+import org.apache.camel.support.jsse.KeyManagersParameters;
+import org.apache.camel.support.jsse.KeyStoreParameters;
+import org.apache.camel.support.jsse.SSLContextParameters;
+import org.apache.camel.support.jsse.SecureSocketProtocolsParameters;
+import org.apache.camel.support.jsse.TrustManagersParameters;
+import org.apache.camel.util.ObjectHelper;
+import org.apache.camel.util.StringHelper;
+import org.apache.kafka.clients.CommonClientConfigs;
+import org.apache.kafka.common.config.SaslConfigs;
+import org.apache.kafka.common.config.SslConfigs;
+import org.apache.kafka.common.config.internals.BrokerSecurityConfigs;
+import org.apache.kafka.common.security.auth.SecurityProtocol;
+
+/**
+ * The options shared by every Kafka client that Camel creates: the brokers, the client id, the connection, metrics and
+ * backoff settings, the security (SSL, SASL, Kerberos, OAuth) settings, the deserializers and header handling, and the
+ * additional properties.
+ * <p/>
+ * A configuration for a specific client extends this class with the options of that client, and builds the client
+ * properties with {@link #applyCommonClientProperties(Properties)}, {@link #applySecurityProperties(Properties)} and
+ * {@link #applyAdditionalProperties(Properties)}.
+ */
+@UriParams
+public abstract class KafkaClientConfiguration implements Cloneable, HeaderFilterStrategyAware {
+
+    @UriParam(label = "common")
+    private String brokers;
+    @UriParam(label = "common")
+    private String clientId;
+    @UriParam(label = "common",
+              description = "To use a custom HeaderFilterStrategy to filter header to and from Camel message.")
+    private HeaderFilterStrategy headerFilterStrategy = new KafkaHeaderFilterStrategy();
+    @UriParam(label = "common", defaultValue = "100")
+    private Integer retryBackoffMs = 100;
+    @UriParam(label = "common", defaultValue = "1000")
+    private Integer retryBackoffMaxMs = 1000;
+    @UriParam(label = "consumer", defaultValue = "true")
+    private boolean preValidateHostAndPort = true;
+    @UriParam(label = "consumer", description = "To use a custom KafkaHeaderDeserializer to deserialize kafka headers values")
+    private KafkaHeaderDeserializer headerDeserializer = new DefaultKafkaHeaderDeserializer();
+    // key.deserializer
+    @UriParam(label = "consumer", defaultValue = KafkaConstants.KAFKA_DEFAULT_DESERIALIZER)
+    private String keyDeserializer = KafkaConstants.KAFKA_DEFAULT_DESERIALIZER;
+    // value.deserializer
+    @UriParam(label = "consumer", defaultValue = KafkaConstants.KAFKA_DEFAULT_DESERIALIZER)
+    private String valueDeserializer = KafkaConstants.KAFKA_DEFAULT_DESERIALIZER;

Review Comment:
   💭 **Design note:** Consumer-specific fields (`preValidateHostAndPort`, `headerDeserializer`, `keyDeserializer`, `valueDeserializer`) live in the abstract base class intended for common client configuration (lines 74–83). While functionally correct (the `@UriParam(label = "consumer")` annotations are preserved), this weakens the semantic contract of `KafkaClientConfiguration` — a future non-consumer Kafka client (admin, streams) extending this class would inherit consumer-only fields.
   
   Consider whether these could remain in `KafkaConfiguration` directly or in a separate consumer mixin, if this base class is intended to be reused for non-consumer Kafka clients in the future.



-- 
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-25029: camel-kafka - extract a reusable client configuration layer [camel]

Posted by "davsclaus (via GitHub)" <gi...@apache.org>.
davsclaus commented on code in PR #27485:
URL: https://github.com/apache/camel/pull/27485#discussion_r4205755477


##########
components/camel-kafka/src/main/java/org/apache/camel/component/kafka/KafkaComponent.java:
##########
@@ -308,6 +213,7 @@ protected void doStart() throws Exception {
         if (kafkaClientFactory == null) {
             kafkaClientFactory = new DefaultKafkaClientFactory();
         }
+

Review Comment:
   Nit (optional): this blank line is unrelated to the refactor and could be dropped to keep the diff minimal.



-- 
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-25029: camel-kafka - extract a reusable client configuration layer [camel]

Posted by "Croway (via GitHub)" <gi...@apache.org>.
Croway merged PR #27485:
URL: https://github.com/apache/camel/pull/27485


-- 
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-25029: camel-kafka - extract a reusable client configuration layer [camel]

Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #27485:
URL: https://github.com/apache/camel/pull/27485#issuecomment-6034956316

   :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-25029: camel-kafka - extract a reusable client configuration layer [camel]

Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #27485:
URL: https://github.com/apache/camel/pull/27485#issuecomment-6035533749

   <!-- ci-tested-modules -->
   :test_tube: **CI tested the following changed modules:**
   
   - `catalog/camel-catalog`
   - `components/camel-kafka`
   - `dsl/camel-componentdsl`
   - `dsl/camel-endpointdsl`
   
   ---
   
   <details><summary>:microscope: Scalpel shadow comparison — Scalpel: 9 of 699 tested, 24 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** (4 direct + 8 downstream), **skip tests for 24** (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/generated/resources/META-INF/org/apache/camel/component/kafka/kafka.json, components/camel-kafka/src/main/java/org/apache/camel/component/kafka/AbstractKafkaComponent.java, components/camel-kafka/src/main/java/org/apache/camel/component/kafka/KafkaClientConfiguration.java, components/camel-kafka/src/main/java/org/apache/camel/component/kafka/KafkaComponent.java, components/camel-kafka/src/main/java/org/apache/camel/component/kafka/KafkaConfiguration.java, components/camel-kafka/src/main/java/org/apache/camel/component/kafka/consumer/support/KafkaRecordProcessor.java, components/camel-kafka/src/main/java/org/apache/camel/component/kafka/security/KafkaSecurityConfigurer.java, components/camel-kafka/src/test/java/org/apache/camel/component/kafka/KafkaClientConfigurationTest.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 (24)</summary>
   
   - `apache-camel`
   - `camel-allcomponents`
   - `camel-catalog-console`
   - `camel-catalog-maven`
   - `camel-catalog-suggest`
   - `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 32s total)</summary>
   
   **Total reactor time:** 5m 32s
   
   | Module | Duration | Status |
   | --- | --- | --- |
   | Camel :: Launcher | 51.2s | SUCCESS |
   | Camel :: JBang :: Plugin :: TUI | 41.7s | SUCCESS |
   | Camel :: JBang :: MCP | 38.7s | SUCCESS |
   | Camel :: Catalog :: Camel Catalog | 23.5s | SUCCESS |
   | Camel :: YAML DSL :: Validator | 18.7s | SUCCESS |
   | Camel :: Component DSL | 18.7s | SUCCESS |
   | Camel :: YAML DSL | 17.1s | SUCCESS |
   | Camel :: JBang :: Plugin :: Validate | 16.6s | SUCCESS |
   | Camel :: Docs | 15.5s | SUCCESS |
   | Camel :: Kamelet Main | 13.1s | SUCCESS |
   | Camel :: JBang :: Plugin :: Testing | 12.1s | SUCCESS |
   | Camel :: JBang :: Plugin :: Kubernetes | 9.7s | SUCCESS |
   | Camel :: Catalog :: Camel Route Parser | 7.7s | SUCCESS |
   | Camel :: YAML DSL :: Deserializers | 7.4s | SUCCESS |
   | Camel :: Catalog :: Camel Report Maven Plugin | 6.4s | SUCCESS |
   | Camel :: All Components Sync point | 6.4s | SUCCESS |
   | Camel :: YAML DSL :: Validator Maven Plugin | 5.1s | SUCCESS |
   | Camel :: YAML DSL :: Maven Plugins | 3.2s | SUCCESS |
   | Camel :: Catalog :: Maven | 3.0s | SUCCESS |
   | Camel :: Catalog :: Suggest (deprecated) | 2.5s | SUCCESS |
   | Camel :: Coverage | 2.0s | SUCCESS |
   | Camel :: Catalog :: Dummy Component | 1.6s | SUCCESS |
   | Camel :: JBang :: Plugin :: Edit | 1.5s | SUCCESS |
   | Camel :: Endpoint DSL :: Support | 1.3s | SUCCESS |
   | Camel :: JBang :: Plugin :: Generate | 1.2s | SUCCESS |
   | Camel :: Assembly | 1.1s | SUCCESS |
   | Camel :: JBang :: Integration tests | 1.0s | SUCCESS |
   | Camel :: JBang :: Plugin :: Route Parser | 0.9s | SUCCESS |
   | Camel :: Catalog :: Console | 0.9s | SUCCESS |
   | Camel :: JBang :: Main | 0.8s | SUCCESS |
   | Camel :: Launcher :: Container | 0.7s | SUCCESS |
   | Camel :: JBang :: Plugin :: MCP | 0.5s | 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` (51.2s)
   - `Camel :: JBang :: Plugin :: TUI` (41.7s)
   - `Camel :: JBang :: MCP` (38.7s)
   - `Camel :: Catalog :: Camel Catalog` (23.5s)
   - `Camel :: YAML DSL :: Validator` (18.7s)
   - `Camel :: Component DSL` (18.7s)
   - `Camel :: YAML DSL` (17.1s)
   - `Camel :: JBang :: Plugin :: Validate` (16.6s)
   - `Camel :: Docs` (15.5s)
   - `Camel :: Kamelet Main` (13.1s)
   - `Camel :: JBang :: Plugin :: Testing` (12.1s)
   - `Camel :: JBang :: Plugin :: Kubernetes` (9.7s)
   - `Camel :: Catalog :: Camel Route Parser` (7.7s)
   - `Camel :: YAML DSL :: Deserializers` (7.4s)
   - `Camel :: Catalog :: Camel Report Maven Plugin` (6.4s)
   - `Camel :: All Components Sync point` (6.4s)
   - `Camel :: YAML DSL :: Validator Maven Plugin` (5.1s)
   - `Camel :: YAML DSL :: Maven Plugins` (3.2s)
   - `Camel :: Catalog :: Maven` (3.0s)
   - `Camel :: Catalog :: Suggest (deprecated)` (2.5s)
   
   </details>
   
   ---
   :gear: [View full build and test results](https://github.com/apache/camel/actions/runs/37600115970)


-- 
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