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: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
