This is an automated email from the ASF dual-hosted git repository.
davsclaus 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 fa91d7f4463d CAMEL-23997: Fix medium-severity findings from
camel-kafka code review
fa91d7f4463d is described below
commit fa91d7f4463d0ce2ea6f4da12ce42e69caed4f53
Author: Claus Ibsen <[email protected]>
AuthorDate: Wed Jul 15 09:05:03 2026 +0200
CAMEL-23997: Fix medium-severity findings from camel-kafka code review
Fix 11 medium-severity findings across configuration, consumer, producer,
and kamelet transforms:
Configuration: sslEndpointAlgorithm=none now correctly disables hostname
verification; offsetRepository properly disables Kafka auto-commit via
consolidated isAutoCommitEnable(); sendBufferBytes applied to consumers
on PLAINTEXT; getAutoCommitEnable() deprecated.
Consumer: reconnect flag properly reset after successful reconnect;
auto-generated groupId shared across all consumer threads; rebalance
listeners migrated to isAutoCommitEnable() fixing offset-repository
manual commit on partition revoke.
Producer batch mode: plain list elements type-converted via configured
serializer; Message keys properly converted; endpoint-configured key
respected for batch elements; per-element header recomputation fixed;
DefaultKafkaHeaderSerializer CamelContext injected in doStart().
Kamelets: ReplaceField renames parsing uses correct guard variable;
MessageTimestampRouter searches for first key present in payload and
handles numeric timestamp fields.
Closes #24693
Co-Authored-By: Claude Opus 4.6 <[email protected]>
---
.../camel/component/kafka/KafkaConfiguration.java | 20 +++---
.../camel/component/kafka/KafkaConsumer.java | 7 +-
.../camel/component/kafka/KafkaFetchRecords.java | 2 +-
.../camel/component/kafka/KafkaProducer.java | 3 +
.../support/classic/ClassicRebalanceListener.java | 2 +-
.../support/resume/ResumeRebalanceListener.java | 2 +-
.../producer/support/KeyValueHolderIterator.java | 25 +++----
.../kafka/transform/MessageTimestampRouter.java | 6 +-
.../component/kafka/transform/ReplaceField.java | 2 +-
.../component/kafka/KafkaConfigurationTest.java | 83 ++++++++++++++++++++++
.../camel/component/kafka/KafkaConsumerTest.java | 29 ++++++++
.../transform/MessageTimestampRouterTest.java | 79 ++++++++++++++++++++
.../kafka/transform/ReplaceFieldTest.java | 14 ++++
.../ROOT/pages/camel-4x-upgrade-guide-4_22.adoc | 22 ++++++
14 files changed, 265 insertions(+), 31 deletions(-)
diff --git
a/components/camel-kafka/src/main/java/org/apache/camel/component/kafka/KafkaConfiguration.java
b/components/camel-kafka/src/main/java/org/apache/camel/component/kafka/KafkaConfiguration.java
index 1e0ec5d2d957..e074ec9e3ad9 100755
---
a/components/camel-kafka/src/main/java/org/apache/camel/component/kafka/KafkaConfiguration.java
+++
b/components/camel-kafka/src/main/java/org/apache/camel/component/kafka/KafkaConfiguration.java
@@ -553,6 +553,8 @@ public class KafkaConfiguration implements Cloneable,
HeaderFilterStrategyAware
String algo = getSslEndpointAlgorithm();
if (algo != null && !algo.equals("none") && !algo.equals("false"))
{
addPropertyIfNotNull(props,
SslConfigs.SSL_ENDPOINT_IDENTIFICATION_ALGORITHM_CONFIG, algo);
+ } else {
+
props.put(SslConfigs.SSL_ENDPOINT_IDENTIFICATION_ALGORITHM_CONFIG, "");
}
addPropertyIfNotEmpty(props,
SslConfigs.SSL_KEYMANAGER_ALGORITHM_CONFIG, getSslKeymanagerAlgorithm());
addPropertyIfNotEmpty(props,
SslConfigs.SSL_TRUSTMANAGER_ALGORITHM_CONFIG, getSslTrustmanagerAlgorithm());
@@ -580,13 +582,14 @@ public class KafkaConfiguration implements Cloneable,
HeaderFilterStrategyAware
addPropertyIfNotEmpty(props,
ConsumerConfig.INTERCEPTOR_CLASSES_CONFIG, getInterceptorClasses());
addPropertyIfNotEmpty(props, ConsumerConfig.AUTO_OFFSET_RESET_CONFIG,
getAutoOffsetReset());
addPropertyIfNotEmpty(props,
ConsumerConfig.CONNECTIONS_MAX_IDLE_MS_CONFIG, getConnectionMaxIdleMs());
- addPropertyIfNotEmpty(props, ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG,
getAutoCommitEnable());
+ addPropertyIfNotEmpty(props, ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG,
isAutoCommitEnable());
if (classicProtocol) {
addPropertyIfNotEmpty(props,
ConsumerConfig.PARTITION_ASSIGNMENT_STRATEGY_CONFIG, getPartitionAssignor());
}
addPropertyIfNotEmpty(props, ConsumerConfig.GROUP_PROTOCOL_CONFIG,
getGroupProtocol());
addPropertyIfNotEmpty(props,
ConsumerConfig.GROUP_REMOTE_ASSIGNOR_CONFIG, getGroupRemoteAssignor());
addPropertyIfNotEmpty(props, ConsumerConfig.RECEIVE_BUFFER_CONFIG,
getReceiveBufferBytes());
+ addPropertyIfNotEmpty(props, ConsumerConfig.SEND_BUFFER_CONFIG,
getSendBufferBytes());
addPropertyIfNotEmpty(props, ConsumerConfig.REQUEST_TIMEOUT_MS_CONFIG,
getConsumerRequestTimeoutMs());
addPropertyIfNotEmpty(props,
ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG, getAutoCommitIntervalMs());
addPropertyIfNotEmpty(props, ConsumerConfig.CHECK_CRCS_CONFIG,
getCheckCrcs());
@@ -641,6 +644,8 @@ public class KafkaConfiguration implements Cloneable,
HeaderFilterStrategyAware
String algo = getSslEndpointAlgorithm();
if (algo != null && !algo.equals("none") && !algo.equals("false"))
{
addPropertyIfNotNull(props,
SslConfigs.SSL_ENDPOINT_IDENTIFICATION_ALGORITHM_CONFIG, algo);
+ } else {
+
props.put(SslConfigs.SSL_ENDPOINT_IDENTIFICATION_ALGORITHM_CONFIG, "");
}
addPropertyIfNotEmpty(props,
SslConfigs.SSL_KEYMANAGER_ALGORITHM_CONFIG, getSslKeymanagerAlgorithm());
addPropertyIfNotEmpty(props,
SslConfigs.SSL_TRUSTMANAGER_ALGORITHM_CONFIG, getSslTrustmanagerAlgorithm());
@@ -649,7 +654,6 @@ public class KafkaConfiguration implements Cloneable,
HeaderFilterStrategyAware
addPropertyIfNotEmpty(props, SslConfigs.SSL_PROTOCOL_CONFIG,
getSslProtocol());
addPropertyIfNotEmpty(props, SslConfigs.SSL_PROVIDER_CONFIG,
getSslProvider());
addUpperCasePropertyIfNotEmpty(props,
SslConfigs.SSL_TRUSTSTORE_TYPE_CONFIG, getSslTruststoreType());
- addPropertyIfNotEmpty(props, ProducerConfig.SEND_BUFFER_CONFIG,
getSendBufferBytes());
}
}
@@ -868,15 +872,15 @@ public class KafkaConfiguration implements Cloneable,
HeaderFilterStrategyAware
}
public boolean isAutoCommitEnable() {
- return offsetRepository == null && autoCommitEnable;
+ return offsetRepository == null && !batching && autoCommitEnable;
}
+ /**
+ * @deprecated use {@link #isAutoCommitEnable()}
+ */
+ @Deprecated(since = "4.22.0")
public boolean getAutoCommitEnable() {
- if (!batching) {
- return autoCommitEnable;
- }
-
- return false;
+ return isAutoCommitEnable();
}
/**
diff --git
a/components/camel-kafka/src/main/java/org/apache/camel/component/kafka/KafkaConsumer.java
b/components/camel-kafka/src/main/java/org/apache/camel/component/kafka/KafkaConsumer.java
index 98f18d942c60..c1bdfe08c240 100644
---
a/components/camel-kafka/src/main/java/org/apache/camel/component/kafka/KafkaConsumer.java
+++
b/components/camel-kafka/src/main/java/org/apache/camel/component/kafka/KafkaConsumer.java
@@ -63,6 +63,7 @@ public class KafkaConsumer extends DefaultConsumer
// This list helps to work around the infinite loop of KAFKA-1894
private final List<KafkaFetchRecords> tasks = new ArrayList<>();
private volatile boolean stopOffsetRepo;
+ private final String defaultGroupId = UUID.randomUUID().toString();
private ResumeStrategy resumeStrategy;
private KafkaConsumerListener consumerListener;
@@ -96,10 +97,6 @@ public class KafkaConsumer extends DefaultConsumer
return (KafkaEndpoint) super.getEndpoint();
}
- private String randomUUID() {
- return UUID.randomUUID().toString();
- }
-
Properties getProps() {
KafkaConfiguration configuration = endpoint.getConfiguration();
@@ -109,7 +106,7 @@ public class KafkaConsumer extends DefaultConsumer
ObjectHelper.ifNotEmpty(endpoint.getKafkaClientFactory().getBrokers(configuration),
v -> props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, v));
- String groupId =
ObjectHelper.supplyIfEmpty(configuration.getGroupId(), this::randomUUID);
+ String groupId = configuration.getGroupId() != null ?
configuration.getGroupId() : defaultGroupId;
props.put(ConsumerConfig.GROUP_ID_CONFIG, groupId);
ObjectHelper.ifNotEmpty(configuration.getGroupInstanceId(),
diff --git
a/components/camel-kafka/src/main/java/org/apache/camel/component/kafka/KafkaFetchRecords.java
b/components/camel-kafka/src/main/java/org/apache/camel/component/kafka/KafkaFetchRecords.java
index cdba2047acec..493fa245fa71 100644
---
a/components/camel-kafka/src/main/java/org/apache/camel/component/kafka/KafkaFetchRecords.java
+++
b/components/camel-kafka/src/main/java/org/apache/camel/component/kafka/KafkaFetchRecords.java
@@ -320,7 +320,7 @@ public class KafkaFetchRecords implements Runnable {
subscribe();
// set reconnect to false as the connection and resume is done at this
point
- setConnected(false);
+ setReconnect(false);
pollExceptionStrategy.reset();
}
diff --git
a/components/camel-kafka/src/main/java/org/apache/camel/component/kafka/KafkaProducer.java
b/components/camel-kafka/src/main/java/org/apache/camel/component/kafka/KafkaProducer.java
index 92e1ab5d8e74..7c8c7d722e66 100755
---
a/components/camel-kafka/src/main/java/org/apache/camel/component/kafka/KafkaProducer.java
+++
b/components/camel-kafka/src/main/java/org/apache/camel/component/kafka/KafkaProducer.java
@@ -29,6 +29,7 @@ import java.util.concurrent.Future;
import com.fasterxml.jackson.databind.JsonNode;
import org.apache.camel.AsyncCallback;
+import org.apache.camel.CamelContextAware;
import org.apache.camel.Exchange;
import org.apache.camel.Message;
import org.apache.camel.component.kafka.producer.support.DelegatingCallback;
@@ -159,6 +160,8 @@ public class KafkaProducer extends DefaultAsyncProducer
implements RouteIdAware
@Override
@SuppressWarnings("rawtypes")
protected void doStart() throws Exception {
+
CamelContextAware.trySetCamelContext(configuration.getHeaderSerializer(),
getEndpoint().getCamelContext());
+
Properties props = getProps();
transactionId =
ObjectHelper.isNotEmpty(configuration.getTransactionalId())
? configuration.getTransactionalId() :
props.getProperty(ProducerConfig.TRANSACTIONAL_ID_CONFIG);
diff --git
a/components/camel-kafka/src/main/java/org/apache/camel/component/kafka/consumer/support/classic/ClassicRebalanceListener.java
b/components/camel-kafka/src/main/java/org/apache/camel/component/kafka/consumer/support/classic/ClassicRebalanceListener.java
index 2a8b3ca49c88..b73ef5bea83f 100644
---
a/components/camel-kafka/src/main/java/org/apache/camel/component/kafka/consumer/support/classic/ClassicRebalanceListener.java
+++
b/components/camel-kafka/src/main/java/org/apache/camel/component/kafka/consumer/support/classic/ClassicRebalanceListener.java
@@ -51,7 +51,7 @@ public class ClassicRebalanceListener implements
ConsumerRebalanceListener {
LOG.debug("onPartitionsRevoked: {} from {}", threadId,
partition.topic());
// only commit offsets if the component has control
- if (!configuration.getAutoCommitEnable()) {
+ if (!configuration.isAutoCommitEnable()) {
commitManager.commit(partition);
}
}
diff --git
a/components/camel-kafka/src/main/java/org/apache/camel/component/kafka/consumer/support/resume/ResumeRebalanceListener.java
b/components/camel-kafka/src/main/java/org/apache/camel/component/kafka/consumer/support/resume/ResumeRebalanceListener.java
index e8847c832387..e32d6c5a2993 100644
---
a/components/camel-kafka/src/main/java/org/apache/camel/component/kafka/consumer/support/resume/ResumeRebalanceListener.java
+++
b/components/camel-kafka/src/main/java/org/apache/camel/component/kafka/consumer/support/resume/ResumeRebalanceListener.java
@@ -52,7 +52,7 @@ public class ResumeRebalanceListener implements
ConsumerRebalanceListener {
LOG.debug("onPartitionsRevoked: {} from {}", threadId,
partition.topic());
// only commit offsets if the component has control
- if (!configuration.getAutoCommitEnable()) {
+ if (!configuration.isAutoCommitEnable()) {
commitManager.commit(partition);
}
}
diff --git
a/components/camel-kafka/src/main/java/org/apache/camel/component/kafka/producer/support/KeyValueHolderIterator.java
b/components/camel-kafka/src/main/java/org/apache/camel/component/kafka/producer/support/KeyValueHolderIterator.java
index 6231cf39e790..8cfb5fbbbe10 100755
---
a/components/camel-kafka/src/main/java/org/apache/camel/component/kafka/producer/support/KeyValueHolderIterator.java
+++
b/components/camel-kafka/src/main/java/org/apache/camel/component/kafka/producer/support/KeyValueHolderIterator.java
@@ -75,10 +75,16 @@ public class KeyValueHolderIterator implements
Iterator<KeyValueHolder<Object, P
propagatedHeadersProvider.getHeaders(ex,
innerMessage)));
}
+ Object convertedBody = tryConvertToSerializedType(exchange, body,
kafkaConfiguration.getValueSerializer());
+ Object key = kafkaConfiguration.getKey() != null
+ ? tryConvertToSerializedType(exchange,
kafkaConfiguration.getKey(), kafkaConfiguration.getKeySerializer())
+ : null;
+
return new KeyValueHolder<>(
body,
new ProducerRecord<>(
- msgTopic, null, null, null, body,
propagatedHeadersProvider.getDefaultHeaders()));
+ msgTopic, null, null, key, convertedBody,
+ propagatedHeadersProvider.getHeaders(exchange,
exchange.getIn())));
}
private Message getInnerMessage(Object object) {
@@ -116,19 +122,14 @@ public class KeyValueHolderIterator implements
Iterator<KeyValueHolder<Object, P
private Object getInnerKey(Exchange innerExchange, Message innerMessage) {
Object innerKey = innerMessage.getHeader(KafkaConstants.KEY);
+ if (innerKey == null) {
+ innerKey = kafkaConfiguration.getKey();
+ }
if (innerKey != null) {
-
- innerKey = kafkaConfiguration.getKey() != null ?
kafkaConfiguration.getKey() : innerKey;
-
- if (innerKey != null) {
- innerKey = tryConvertToSerializedType(innerExchange, innerKey,
- kafkaConfiguration.getKeySerializer());
- }
-
- return innerKey;
+ Exchange ex = innerExchange != null ? innerExchange : exchange;
+ innerKey = tryConvertToSerializedType(ex, innerKey,
kafkaConfiguration.getKeySerializer());
}
-
- return null;
+ return innerKey;
}
private Integer getInnerPartitionKey(Message innerMessage) {
diff --git
a/components/camel-kafka/src/main/java/org/apache/camel/component/kafka/transform/MessageTimestampRouter.java
b/components/camel-kafka/src/main/java/org/apache/camel/component/kafka/transform/MessageTimestampRouter.java
index 9009188d9a20..6225f938392d 100644
---
a/components/camel-kafka/src/main/java/org/apache/camel/component/kafka/transform/MessageTimestampRouter.java
+++
b/components/camel-kafka/src/main/java/org/apache/camel/component/kafka/transform/MessageTimestampRouter.java
@@ -59,7 +59,9 @@ public class MessageTimestampRouter {
for (String key : splittedKeys) {
if (ObjectHelper.isNotEmpty(key)) {
rawTimestamp = body.get(key);
- break;
+ if (rawTimestamp != null) {
+ break;
+ }
}
}
Long timestamp = null;
@@ -67,7 +69,7 @@ public class MessageTimestampRouter {
&& !timestampKeyFormat.equalsIgnoreCase("timestamp")) {
final SimpleDateFormat timestampKeyFmt = new
SimpleDateFormat(timestampKeyFormat);
timestampKeyFmt.setTimeZone(TimeZone.getTimeZone("UTC"));
- timestamp = timestampKeyFmt.parse((String) rawTimestamp).getTime();
+ timestamp =
timestampKeyFmt.parse(rawTimestamp.toString()).getTime();
} else if (ObjectHelper.isNotEmpty(rawTimestamp)) {
timestamp = Long.parseLong(rawTimestamp.toString());
}
diff --git
a/components/camel-kafka/src/main/java/org/apache/camel/component/kafka/transform/ReplaceField.java
b/components/camel-kafka/src/main/java/org/apache/camel/component/kafka/transform/ReplaceField.java
index 1d225f17e0ba..c9d0499373af 100644
---
a/components/camel-kafka/src/main/java/org/apache/camel/component/kafka/transform/ReplaceField.java
+++
b/components/camel-kafka/src/main/java/org/apache/camel/component/kafka/transform/ReplaceField.java
@@ -46,7 +46,7 @@ public class ReplaceField {
if (ObjectHelper.isNotEmpty(disabled) &&
!disabled.equalsIgnoreCase("none")) {
disabledFields =
Arrays.stream(disabled.split(",")).collect(Collectors.toList());
}
- if (ObjectHelper.isNotEmpty(disabled)) {
+ if (ObjectHelper.isNotEmpty(renames)) {
renameFields =
Arrays.stream(renames.split(",")).collect(Collectors.toList());
}
Map<Object, Object> updatedBody = new HashMap<>();
diff --git
a/components/camel-kafka/src/test/java/org/apache/camel/component/kafka/KafkaConfigurationTest.java
b/components/camel-kafka/src/test/java/org/apache/camel/component/kafka/KafkaConfigurationTest.java
new file mode 100644
index 000000000000..0f211726799b
--- /dev/null
+++
b/components/camel-kafka/src/test/java/org/apache/camel/component/kafka/KafkaConfigurationTest.java
@@ -0,0 +1,83 @@
+/*
+ * 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.Properties;
+
+import org.apache.camel.spi.StateRepository;
+import org.apache.kafka.clients.consumer.ConsumerConfig;
+import org.apache.kafka.common.config.SslConfigs;
+import org.junit.jupiter.api.Test;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.mockito.Mockito.mock;
+
+class KafkaConfigurationTest {
+
+ @Test
+ void sslEndpointAlgorithmNoneDisablesHostnameVerification() {
+ KafkaConfiguration config = new KafkaConfiguration();
+ config.setBrokers("localhost:9092");
+ config.setSecurityProtocol("SSL");
+ config.setSslEndpointAlgorithm("none");
+
+ Properties props = config.createProducerProperties();
+ assertEquals("",
props.get(SslConfigs.SSL_ENDPOINT_IDENTIFICATION_ALGORITHM_CONFIG));
+ }
+
+ @Test
+ void sslEndpointAlgorithmFalseDisablesHostnameVerification() {
+ KafkaConfiguration config = new KafkaConfiguration();
+ config.setBrokers("localhost:9092");
+ config.setSecurityProtocol("SSL");
+ config.setSslEndpointAlgorithm("false");
+
+ Properties consumerProps = config.createConsumerProperties();
+ assertEquals("",
consumerProps.get(SslConfigs.SSL_ENDPOINT_IDENTIFICATION_ALGORITHM_CONFIG));
+ }
+
+ @Test
+ @SuppressWarnings("unchecked")
+ void offsetRepositoryDisablesAutoCommit() {
+ KafkaConfiguration config = new KafkaConfiguration();
+ config.setBrokers("localhost:9092");
+ config.setOffsetRepository(mock(StateRepository.class));
+
+ Properties props = config.createConsumerProperties();
+ assertEquals(false,
props.get(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG));
+ }
+
+ @Test
+ void batchingDisablesAutoCommit() {
+ KafkaConfiguration config = new KafkaConfiguration();
+ config.setBrokers("localhost:9092");
+ config.setBatching(true);
+
+ Properties props = config.createConsumerProperties();
+ assertEquals(false,
props.get(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG));
+ }
+
+ @Test
+ void sendBufferBytesAppliedToConsumerWithoutSsl() {
+ KafkaConfiguration config = new KafkaConfiguration();
+ config.setBrokers("localhost:9092");
+ config.setSendBufferBytes(131072);
+
+ Properties props = config.createConsumerProperties();
+ assertEquals(131072, props.get(ConsumerConfig.SEND_BUFFER_CONFIG));
+ }
+}
diff --git
a/components/camel-kafka/src/test/java/org/apache/camel/component/kafka/KafkaConsumerTest.java
b/components/camel-kafka/src/test/java/org/apache/camel/component/kafka/KafkaConsumerTest.java
index 58f95e07fda4..4f10f997f8ad 100644
---
a/components/camel-kafka/src/test/java/org/apache/camel/component/kafka/KafkaConsumerTest.java
+++
b/components/camel-kafka/src/test/java/org/apache/camel/component/kafka/KafkaConsumerTest.java
@@ -16,13 +16,18 @@
*/
package org.apache.camel.component.kafka;
+import java.util.Properties;
+
import org.apache.camel.CamelContext;
import org.apache.camel.ExtendedCamelContext;
import org.apache.camel.Processor;
import org.apache.camel.spi.ExchangeFactory;
+import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.junit.jupiter.api.Test;
import static org.junit.jupiter.api.Assertions.assertDoesNotThrow;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
import static org.junit.jupiter.api.Assertions.assertThrows;
import static org.mockito.ArgumentMatchers.any;
import static org.mockito.Mockito.mock;
@@ -56,6 +61,30 @@ public class KafkaConsumerTest {
assertThrows(IllegalArgumentException.class, () ->
kafkaConsumer.getProps());
}
+ @Test
+ public void autoGeneratedGroupIdIsSharedAcrossCalls() {
+ when(endpoint.getCamelContext()).thenReturn(context);
+ when(context.getCamelContextExtension()).thenReturn(ecc);
+ when(ecc.getExchangeFactory()).thenReturn(ef);
+ when(ef.newExchangeFactory(any())).thenReturn(ef);
+ when(endpoint.getComponent()).thenReturn(component);
+ when(endpoint.getConfiguration()).thenReturn(configuration);
+ when(configuration.getGroupId()).thenReturn(null);
+ when(configuration.createConsumerProperties()).thenReturn(new
Properties());
+ when(endpoint.getKafkaClientFactory()).thenReturn(clientFactory);
+ when(clientFactory.getBrokers(any())).thenReturn("localhost:9092");
+ when(component.getKafkaClientFactory()).thenReturn(clientFactory);
+ final KafkaConsumer kafkaConsumer = new KafkaConsumer(endpoint,
processor);
+
+ Properties props1 = kafkaConsumer.getProps();
+ Properties props2 = kafkaConsumer.getProps();
+
+ String groupId1 = (String) props1.get(ConsumerConfig.GROUP_ID_CONFIG);
+ String groupId2 = (String) props2.get(ConsumerConfig.GROUP_ID_CONFIG);
+ assertNotNull(groupId1);
+ assertEquals(groupId1, groupId2);
+ }
+
@Test
public void consumerOnlyRequiresBootstrapServers() {
when(endpoint.getCamelContext()).thenReturn(context);
diff --git
a/components/camel-kafka/src/test/java/org/apache/camel/component/kafka/transform/MessageTimestampRouterTest.java
b/components/camel-kafka/src/test/java/org/apache/camel/component/kafka/transform/MessageTimestampRouterTest.java
new file mode 100644
index 000000000000..5130e64df2fa
--- /dev/null
+++
b/components/camel-kafka/src/test/java/org/apache/camel/component/kafka/transform/MessageTimestampRouterTest.java
@@ -0,0 +1,79 @@
+/*
+ * 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.transform;
+
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.fasterxml.jackson.databind.node.ObjectNode;
+import org.apache.camel.Exchange;
+import org.apache.camel.component.kafka.KafkaConstants;
+import org.apache.camel.impl.DefaultCamelContext;
+import org.apache.camel.support.DefaultExchange;
+import org.junit.jupiter.api.Test;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertNull;
+
+class MessageTimestampRouterTest {
+
+ private final MessageTimestampRouter router = new MessageTimestampRouter();
+ private final ObjectMapper mapper = new ObjectMapper();
+
+ @Test
+ void shouldFallbackToSecondKeyWhenFirstKeyMissing() throws Exception {
+ DefaultCamelContext camelContext = new DefaultCamelContext();
+ Exchange exchange = new DefaultExchange(camelContext);
+ exchange.getMessage().setHeader(KafkaConstants.TOPIC, "my-topic");
+
+ ObjectNode body = mapper.createObjectNode();
+ body.put("ts2", "1719100800000");
+ exchange.getMessage().setBody(body);
+
+ router.process("$[topic]_$[timestamp]", "yyyy-MM-dd", "ts1,ts2",
"timestamp", exchange);
+
+ assertEquals("my-topic_2024-06-23",
exchange.getMessage().getHeader(KafkaConstants.OVERRIDE_TOPIC));
+ }
+
+ @Test
+ void shouldHandleNumericTimestampField() throws Exception {
+ DefaultCamelContext camelContext = new DefaultCamelContext();
+ Exchange exchange = new DefaultExchange(camelContext);
+ exchange.getMessage().setHeader(KafkaConstants.TOPIC, "my-topic");
+
+ ObjectNode body = mapper.createObjectNode();
+ body.put("ts", 1719100800000L);
+ exchange.getMessage().setBody(body);
+
+ router.process("$[topic]_$[timestamp]", "yyyy-MM-dd", "ts",
"timestamp", exchange);
+
+ assertEquals("my-topic_2024-06-23",
exchange.getMessage().getHeader(KafkaConstants.OVERRIDE_TOPIC));
+ }
+
+ @Test
+ void shouldNotSetTopicWhenNoTimestampKeyMatches() throws Exception {
+ DefaultCamelContext camelContext = new DefaultCamelContext();
+ Exchange exchange = new DefaultExchange(camelContext);
+ exchange.getMessage().setHeader(KafkaConstants.TOPIC, "my-topic");
+
+ ObjectNode body = mapper.createObjectNode();
+ body.put("other", "value");
+ exchange.getMessage().setBody(body);
+
+ router.process("$[topic]_$[timestamp]", "yyyy-MM-dd", "ts1,ts2",
"timestamp", exchange);
+
+
assertNull(exchange.getMessage().getHeader(KafkaConstants.OVERRIDE_TOPIC));
+ }
+}
diff --git
a/components/camel-kafka/src/test/java/org/apache/camel/component/kafka/transform/ReplaceFieldTest.java
b/components/camel-kafka/src/test/java/org/apache/camel/component/kafka/transform/ReplaceFieldTest.java
index b42aceafd99d..1ad133b84dc0 100644
---
a/components/camel-kafka/src/test/java/org/apache/camel/component/kafka/transform/ReplaceFieldTest.java
+++
b/components/camel-kafka/src/test/java/org/apache/camel/component/kafka/transform/ReplaceFieldTest.java
@@ -100,6 +100,20 @@ class ReplaceFieldTest {
"}");
}
+ @Test
+ void shouldRenameFieldsWithoutDisabledSet() throws Exception {
+ Exchange exchange = new DefaultExchange(camelContext);
+
+ exchange.getMessage().setBody(mapper.readTree(baseJson));
+
+ JsonNode node = processor.process("all", null,
"name:firstName,age:years", exchange);
+
+ Assertions.assertEquals(node.toString(), "{" +
+ "\"firstName\":\"Rajesh
Koothrappali\"," +
+ "\"years\":\"29\"" +
+ "}");
+ }
+
@Test
void shouldReplaceFieldWithDisableAllFields() throws Exception {
Exchange exchange = new DefaultExchange(camelContext);
diff --git
a/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_22.adoc
b/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_22.adoc
index 935ef84e5273..ab991634201e 100644
--- a/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_22.adoc
+++ b/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_22.adoc
@@ -267,6 +267,28 @@ Scalar value nodes are now sent as a single record. Only
container nodes (`Array
are still split, which is unchanged. Any `convertBodyTo(...)` previously used
as a workaround is no
longer required.
+=== camel-kafka - sslEndpointAlgorithm=none now disables hostname verification
+
+Setting `sslEndpointAlgorithm` to `none` or `false` now correctly disables SSL
hostname verification
+by setting `ssl.endpoint.identification.algorithm` to an empty string.
Previously the property was
+simply omitted, which caused Kafka clients to fall back to their default
(`https`), leaving hostname
+verification silently enabled.
+
+=== camel-kafka - Auto-generated groupId shared across consumer threads
+
+When no `groupId` is configured, the auto-generated UUID is now shared across
all consumer threads
+(`consumersCount`). Previously each thread received its own random UUID,
causing each thread to
+independently consume all partitions, which resulted in every message being
processed `consumersCount`
+times.
+
+=== camel-kafka - Batch producer respects endpoint key for plain list elements
+
+The batch producer (when the body is a `List`) now respects the
endpoint-configured `key` option for
+elements that do not have a `CamelKafkaKey` header. Previously, elements
without the header would
+get a null record key even when `key=fixedKey` was configured on the endpoint.
Additionally, plain
+list elements (not `Exchange` or `Message`) are now properly converted to the
configured serializer
+type.
+
=== camel-jbang catalog tables fill the terminal width
The `camel catalog` commands (`camel catalog component`, `camel catalog
dataformat`,
