This is an automated email from the ASF dual-hosted git repository. davsclaus pushed a commit to branch fix/CAMEL-23997 in repository https://gitbox.apache.org/repos/asf/camel.git
commit 237f577f131fde658bedd9f903fbf26dc58ffed7 Author: Claus Ibsen <[email protected]> AuthorDate: Tue Jul 14 20:51:50 2026 +0200 CAMEL-23997: Fix medium-severity findings from camel-kafka code review Co-Authored-By: Claude Opus 4.6 <[email protected]> Signed-off-by: Claus Ibsen <[email protected]> --- .../camel/component/kafka/KafkaConfiguration.java | 8 ++- .../camel/component/kafka/KafkaConsumer.java | 7 +- .../camel/component/kafka/KafkaFetchRecords.java | 2 +- .../camel/component/kafka/KafkaProducer.java | 3 + .../producer/support/KeyValueHolderIterator.java | 25 +++---- .../kafka/transform/MessageTimestampRouter.java | 6 +- .../component/kafka/transform/ReplaceField.java | 2 +- .../component/kafka/KafkaConfigurationTest.java | 73 ++++++++++++++++++++ .../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 ++++++ 12 files changed, 247 insertions(+), 23 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 57372b08af62..0b2d9246653b 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 @@ -551,6 +551,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()); @@ -578,13 +580,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()); @@ -639,6 +642,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()); @@ -647,7 +652,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()); } } 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/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..21b64103f291 --- /dev/null +++ b/components/camel-kafka/src/test/java/org/apache/camel/component/kafka/KafkaConfigurationTest.java @@ -0,0 +1,73 @@ +/* + * 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 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 0454ac8a835a..6194df29bf58 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 @@ -243,6 +243,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`,
