This is an automated email from the ASF dual-hosted git repository.
oscerd pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/camel-kafka-connector.git
The following commit(s) were added to refs/heads/main by this push:
new 3e68d3022b Fix #1497: pass Kafka client properties to the Kafka
idempotent repository (#1866)
3e68d3022b is described below
commit 3e68d3022bbb1b1f8fb6a49cdecc916182c98d82
Author: Andrea Cosentino <[email protected]>
AuthorDate: Thu Nov 5 13:08:34 2026 +0100
Fix #1497: pass Kafka client properties to the Kafka idempotent repository
(#1866)
The Kafka idempotent repository was built from
camel.idempotency.kafka.bootstrap.servers only, so it could not reach a
cluster secured with SSL or SASL.
- Add the camel.idempotency.kafka.client.<property> prefix: the
properties under it, with the prefix removed, are applied to the
consumer and the producer of the repository, on top of the bootstrap
servers. Without them the repository is built as before.
- Mask every value under the prefix in the startup properties log:
SensitiveUtils only checks the last segment of a key, so
sasl.jaas.config or ssl.keystore.key would have been logged in clear.
- Document the prefix, with SSL and SASL examples that read the secrets
through a Kafka Connect config provider.
Signed-off-by: Andrea Cosentino <[email protected]>
Co-authored-by: Claude Opus 5.5 <[email protected]>
---
.../camel/kafkaconnector/CamelConnectorConfig.java | 7 ++
.../apache/camel/kafkaconnector/CamelSinkTask.java | 2 +
.../camel/kafkaconnector/CamelSourceTask.java | 2 +
.../utils/CamelKafkaConnectMain.java | 49 +++++++++-
.../camel/kafkaconnector/CamelSinkTaskTest.java | 20 ++++
.../camel/kafkaconnector/CamelSourceTaskTest.java | 27 ++++++
...mpotentRepositoryKafkaClientPropertiesTest.java | 107 +++++++++++++++++++++
.../utils/StartupPropertiesLoggingTest.java | 84 ++++++++++++++++
.../ROOT/pages/user-guide/basic-configuration.adoc | 2 +
.../modules/ROOT/pages/user-guide/idempotency.adoc | 55 +++++++++++
10 files changed, 353 insertions(+), 2 deletions(-)
diff --git
a/core/src/main/java/org/apache/camel/kafkaconnector/CamelConnectorConfig.java
b/core/src/main/java/org/apache/camel/kafkaconnector/CamelConnectorConfig.java
index 98980bf419..cee3ae360f 100644
---
a/core/src/main/java/org/apache/camel/kafkaconnector/CamelConnectorConfig.java
+++
b/core/src/main/java/org/apache/camel/kafkaconnector/CamelConnectorConfig.java
@@ -90,6 +90,13 @@ public abstract class CamelConnectorConfig extends
AbstractConfig {
public static final String
CAMEL_CONNECTOR_IDEMPOTENCY_KAFKA_BOOTSTRAP_SERVERS_CONF =
"camel.idempotency.kafka.bootstrap.servers";
public static final String
CAMEL_CONNECTOR_IDEMPOTENCY_KAFKA_BOOTSTRAP_SERVERS_DOC = "A comma-separated
list of host and port pairs that are the addresses of the Kafka brokers where
the idempotent repository should live";
+ // A prefix rather than a single option, so like camel.source.endpoint. it
has no ConfigDef entry and no default:
+ // every key that starts with it is a Kafka client property for the Kafka
idempotent repository.
+ public static final String CAMEL_CONNECTOR_IDEMPOTENCY_KAFKA_CLIENT_PREFIX
= "camel.idempotency.kafka.client.";
+ public static final String
CAMEL_CONNECTOR_IDEMPOTENCY_KAFKA_CLIENT_PREFIX_DOC = "Prefix for Kafka client
properties to pass, with the prefix removed, to the consumer"
+ + " and the producer of the Kafka idempotent repository, on top of
" + CAMEL_CONNECTOR_IDEMPOTENCY_KAFKA_BOOTSTRAP_SERVERS_CONF
+ + ", for example " +
CAMEL_CONNECTOR_IDEMPOTENCY_KAFKA_CLIENT_PREFIX + "security.protocol=SSL. Use
it for the SSL or SASL settings of a secured cluster";
+
public static final int
CAMEL_CONNECTOR_IDEMPOTENCY_KAFKA_MAX_CACHE_SIZE_DEFAULT = 1000;
public static final String
CAMEL_CONNECTOR_IDEMPOTENCY_KAFKA_MAX_CACHE_SIZE_CONF =
"camel.idempotency.kafka.max.cache.size";
public static final String
CAMEL_CONNECTOR_IDEMPOTENCY_KAFKA_MAX_CACHE_SIZE_DOC = "Sets the maximum size
of the local key cache";
diff --git
a/core/src/main/java/org/apache/camel/kafkaconnector/CamelSinkTask.java
b/core/src/main/java/org/apache/camel/kafkaconnector/CamelSinkTask.java
index cacbc3969c..77795ec467 100644
--- a/core/src/main/java/org/apache/camel/kafkaconnector/CamelSinkTask.java
+++ b/core/src/main/java/org/apache/camel/kafkaconnector/CamelSinkTask.java
@@ -125,6 +125,7 @@ public class CamelSinkTask extends SinkTask {
final String idempotentRepositoryBootstrapServers =
config.getString(CamelSinkConnectorConfig.CAMEL_CONNECTOR_IDEMPOTENCY_KAFKA_BOOTSTRAP_SERVERS_CONF);
final int idempotentRepositoryKafkaMaxCacheSize =
config.getInt(CamelSinkConnectorConfig.CAMEL_CONNECTOR_IDEMPOTENCY_KAFKA_MAX_CACHE_SIZE_CONF);
final int idempotentRepositoryKafkaPollDuration =
config.getInt(CamelSinkConnectorConfig.CAMEL_CONNECTOR_IDEMPOTENCY_KAFKA_POLL_DURATION_CONF);
+ final Map<String, Object>
idempotentRepositoryKafkaClientProperties =
config.originalsWithPrefix(CamelSinkConnectorConfig.CAMEL_CONNECTOR_IDEMPOTENCY_KAFKA_CLIENT_PREFIX);
final String headersRemovePattern =
config.getString(CamelSinkConnectorConfig.CAMEL_CONNECTOR_REMOVE_HEADERS_PATTERN_CONF);
mapProperties =
config.getBoolean(CamelSinkConnectorConfig.CAMEL_CONNECTOR_MAP_PROPERTIES_CONF);
mapHeaders =
config.getBoolean(CamelSinkConnectorConfig.CAMEL_CONNECTOR_MAP_HEADERS_CONF);
@@ -168,6 +169,7 @@ public class CamelSinkTask extends SinkTask {
.withIdempotentRepositoryKafkaServers(idempotentRepositoryBootstrapServers)
.withIdempotentRepositoryKafkaMaxCacheSize(idempotentRepositoryKafkaMaxCacheSize)
.withIdempotentRepositoryKafkaPollDuration(idempotentRepositoryKafkaPollDuration)
+
.withIdempotentRepositoryKafkaClientProperties(idempotentRepositoryKafkaClientProperties)
.withHeadersExcludePattern(headersRemovePattern)
.withDumpRoutes(dumpRoutes)
.build(camelContext);
diff --git
a/core/src/main/java/org/apache/camel/kafkaconnector/CamelSourceTask.java
b/core/src/main/java/org/apache/camel/kafkaconnector/CamelSourceTask.java
index 27dcbc8f30..218bd17c1a 100644
--- a/core/src/main/java/org/apache/camel/kafkaconnector/CamelSourceTask.java
+++ b/core/src/main/java/org/apache/camel/kafkaconnector/CamelSourceTask.java
@@ -125,6 +125,7 @@ public class CamelSourceTask extends SourceTask {
final String idempotentRepositoryBootstrapServers =
config.getString(CamelSourceConnectorConfig.CAMEL_CONNECTOR_IDEMPOTENCY_KAFKA_BOOTSTRAP_SERVERS_CONF);
final int idempotentRepositoryKafkaMaxCacheSize =
config.getInt(CamelSourceConnectorConfig.CAMEL_CONNECTOR_IDEMPOTENCY_KAFKA_MAX_CACHE_SIZE_CONF);
final int idempotentRepositoryKafkaPollDuration =
config.getInt(CamelSourceConnectorConfig.CAMEL_CONNECTOR_IDEMPOTENCY_KAFKA_POLL_DURATION_CONF);
+ final Map<String, Object>
idempotentRepositoryKafkaClientProperties =
config.originalsWithPrefix(CamelSourceConnectorConfig.CAMEL_CONNECTOR_IDEMPOTENCY_KAFKA_CLIENT_PREFIX);
final String headersRemovePattern =
config.getString(CamelSourceConnectorConfig.CAMEL_CONNECTOR_REMOVE_HEADERS_PATTERN_CONF);
mapProperties =
config.getBoolean(CamelSourceConnectorConfig.CAMEL_CONNECTOR_MAP_PROPERTIES_CONF);
mapHeaders =
config.getBoolean(CamelSinkConnectorConfig.CAMEL_CONNECTOR_MAP_HEADERS_CONF);
@@ -185,6 +186,7 @@ public class CamelSourceTask extends SourceTask {
.withIdempotentRepositoryKafkaServers(idempotentRepositoryBootstrapServers)
.withIdempotentRepositoryKafkaMaxCacheSize(idempotentRepositoryKafkaMaxCacheSize)
.withIdempotentRepositoryKafkaPollDuration(idempotentRepositoryKafkaPollDuration)
+
.withIdempotentRepositoryKafkaClientProperties(idempotentRepositoryKafkaClientProperties)
.withHeadersExcludePattern(headersRemovePattern)
.withDumpRoutes(dumpRoutes)
.build(camelContext);
diff --git
a/core/src/main/java/org/apache/camel/kafkaconnector/utils/CamelKafkaConnectMain.java
b/core/src/main/java/org/apache/camel/kafkaconnector/utils/CamelKafkaConnectMain.java
index 183e9ceada..844477b4a9 100644
---
a/core/src/main/java/org/apache/camel/kafkaconnector/utils/CamelKafkaConnectMain.java
+++
b/core/src/main/java/org/apache/camel/kafkaconnector/utils/CamelKafkaConnectMain.java
@@ -16,6 +16,7 @@
*/
package org.apache.camel.kafkaconnector.utils;
+import java.util.Collections;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
@@ -40,6 +41,7 @@ import org.apache.camel.support.service.ServiceHelper;
import org.apache.camel.util.ObjectHelper;
import org.apache.camel.util.SensitiveUtils;
import org.apache.camel.util.URISupport;
+import org.apache.kafka.clients.CommonClientConfigs;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -120,6 +122,7 @@ public class CamelKafkaConnectMain extends SimpleMain {
private String idempotentRepositoryKafkaServers;
private int idempotentRepositoryKafkaMaxCacheSize;
private int idempotentRepositoryKafkaPollDuration;
+ private Map<String, Object> idempotentRepositoryKafkaClientProperties
= Collections.emptyMap();
private String headersExcludePattern;
private boolean removeHeadersFirst;
private UnaryOperator<AggregationStrategy>
aggregationStrategyDecorator;
@@ -215,6 +218,19 @@ public class CamelKafkaConnectMain extends SimpleMain {
return this;
}
+ /**
+ * Kafka client properties (security.protocol, sasl.jaas.config,
ssl.truststore.location, ...) for the consumer
+ * and the producer of the Kafka idempotent repository, for example to
reach a cluster secured with SSL or
+ * SASL. They are applied on top of the bootstrap servers.
+ *
+ * @param idempotentRepositoryKafkaClientProperties the Kafka client
properties, keyed by their Kafka name.
+ */
+ public Builder
withIdempotentRepositoryKafkaClientProperties(Map<String, ?>
idempotentRepositoryKafkaClientProperties) {
+ this.idempotentRepositoryKafkaClientProperties =
idempotentRepositoryKafkaClientProperties == null
+ ? Collections.emptyMap() : new
HashMap<>(idempotentRepositoryKafkaClientProperties);
+ return this;
+ }
+
public Builder withHeadersExcludePattern(String headersExcludePattern)
{
this.headersExcludePattern = headersExcludePattern;
return this;
@@ -255,7 +271,12 @@ public class CamelKafkaConnectMain extends SimpleMain {
private String filterSensitive(Map.Entry<Object, Object> entry) {
final String key = (String) entry.getKey();
- if (SensitiveUtils.containsSensitive(key)) {
+ // The Kafka client properties of the idempotent repository are
masked whatever their name. SensitiveUtils
+ // only checks the last segment of a key, which misses secrets
such as sasl.jaas.config or ssl.keystore.key,
+ // and the options of a custom login module or callback handler
can be called anything. The Kafka clients
+ // log the configuration they use themselves, with their own
password masking.
+ if (SensitiveUtils.containsSensitive(key)
+ ||
key.startsWith(CamelConnectorConfig.CAMEL_CONNECTOR_IDEMPOTENCY_KAFKA_CLIENT_PREFIX))
{
return key + "=xxxxxxx";
}
final Object value = entry.getValue();
@@ -269,6 +290,30 @@ public class CamelKafkaConnectMain extends SimpleMain {
return key + "=" + value;
}
+ private KafkaIdempotentRepository createKafkaIdempotentRepository() {
+ if (idempotentRepositoryKafkaClientProperties.isEmpty()) {
+ return new
KafkaIdempotentRepository(idempotentRepositoryTopicName,
idempotentRepositoryKafkaServers,
+ idempotentRepositoryKafkaMaxCacheSize,
idempotentRepositoryKafkaPollDuration);
+ }
+ // Given a consumer and a producer configuration the repository no
longer reads its bootstrap servers, so
+ // they go into both. It still sets its own String (de)serializers
and disables auto commit when it starts.
+ return new KafkaIdempotentRepository(idempotentRepositoryTopicName,
+ idempotentRepositoryKafkaClientConfig(),
idempotentRepositoryKafkaClientConfig(),
+ idempotentRepositoryKafkaMaxCacheSize,
idempotentRepositoryKafkaPollDuration);
+ }
+
+ // A new instance on each call: when it starts, the repository adds
consumer-only settings to its consumer
+ // configuration and producer-only settings to its producer
configuration.
+ private Properties idempotentRepositoryKafkaClientConfig() {
+ Properties config = new Properties();
+ if (idempotentRepositoryKafkaServers != null) {
+ config.put(CommonClientConfigs.BOOTSTRAP_SERVERS_CONFIG,
idempotentRepositoryKafkaServers);
+ }
+ // applied last, so a bootstrap.servers given as a client property
wins
+ config.putAll(idempotentRepositoryKafkaClientProperties);
+ return config;
+ }
+
public CamelKafkaConnectMain build(CamelContext camelContext) {
CamelKafkaConnectMain camelMain = new
CamelKafkaConnectMain(camelContext);
camelMain.configure().setAutoConfigurationLogSummary(false);
@@ -330,7 +375,7 @@ public class CamelKafkaConnectMain extends SimpleMain {
idempotentRepo =
MemoryIdempotentRepository.memoryIdempotentRepository(memoryDimension);
break;
case "kafka":
- idempotentRepo = new
KafkaIdempotentRepository(idempotentRepositoryTopicName,
idempotentRepositoryKafkaServers, idempotentRepositoryKafkaMaxCacheSize,
idempotentRepositoryKafkaPollDuration);
+ idempotentRepo = createKafkaIdempotentRepository();
break;
default:
break;
diff --git
a/core/src/test/java/org/apache/camel/kafkaconnector/CamelSinkTaskTest.java
b/core/src/test/java/org/apache/camel/kafkaconnector/CamelSinkTaskTest.java
index be77a123bd..9213cbae59 100644
--- a/core/src/test/java/org/apache/camel/kafkaconnector/CamelSinkTaskTest.java
+++ b/core/src/test/java/org/apache/camel/kafkaconnector/CamelSinkTaskTest.java
@@ -1017,6 +1017,26 @@ class CamelSinkTaskTest {
.getString(CamelSinkConnectorConfig.CAMEL_SINK_CONTENT_LOG_LEVEL_CONF));
}
+ @Test
+ void testKafkaIdempotentRepositoryGetsTheClientProperties() {
+ final String clientPrefix =
CamelSinkConnectorConfig.CAMEL_CONNECTOR_IDEMPOTENCY_KAFKA_CLIENT_PREFIX;
+ final String truststore = "target/" + getClass().getSimpleName() +
"-missing-truststore.jks";
+ Map<String, String> props = new HashMap<>();
+ props.put(TOPIC_CONF, TOPIC_NAME);
+ props.put(CamelSinkConnectorConfig.CAMEL_SINK_URL_CONF, SEDA_URI);
+
props.put(CamelSinkConnectorConfig.CAMEL_CONNECTOR_IDEMPOTENCY_ENABLED_CONF,
"true");
+
props.put(CamelSinkConnectorConfig.CAMEL_CONNECTOR_IDEMPOTENCY_REPOSITORY_TYPE_CONF,
"kafka");
+ props.put(clientPrefix + "security.protocol", "SSL");
+ props.put(clientPrefix + "ssl.truststore.location", truststore);
+ props.put(clientPrefix + "ssl.truststore.password", "changeit");
+
+ // No broker needed: the truststore does not exist, so the repository
fails as soon as it creates its Kafka
+ // consumer ("Failed to load SSL keystore <path>"). That only happens
if the consumer was given the SSL
+ // settings under their Kafka names.
+ ConnectException e = assertThrows(ConnectException.class, () ->
sinkTask.start(props));
+ assertThat(e).hasStackTraceContaining(truststore);
+ }
+
@Test
void testWithIdempotencyAndHeader() throws InterruptedException {
Map<String, String> props = new HashMap<>();
diff --git
a/core/src/test/java/org/apache/camel/kafkaconnector/CamelSourceTaskTest.java
b/core/src/test/java/org/apache/camel/kafkaconnector/CamelSourceTaskTest.java
index e7a771d990..38e929adac 100644
---
a/core/src/test/java/org/apache/camel/kafkaconnector/CamelSourceTaskTest.java
+++
b/core/src/test/java/org/apache/camel/kafkaconnector/CamelSourceTaskTest.java
@@ -41,6 +41,7 @@ import
org.apache.camel.kafkaconnector.utils.StringJoinerAggregator;
import org.apache.camel.support.SynchronizationAdapter;
import org.apache.kafka.connect.data.Decimal;
import org.apache.kafka.connect.data.Schema;
+import org.apache.kafka.connect.errors.ConnectException;
import org.apache.kafka.connect.header.Header;
import org.apache.kafka.connect.json.JsonConverter;
import org.apache.kafka.connect.json.JsonConverterConfig;
@@ -56,6 +57,7 @@ import static
org.junit.jupiter.api.Assertions.assertNotEquals;
import static org.junit.jupiter.api.Assertions.assertNotNull;
import static org.junit.jupiter.api.Assertions.assertNull;
import static org.junit.jupiter.api.Assertions.assertSame;
+import static org.junit.jupiter.api.Assertions.assertThrows;
import static org.junit.jupiter.api.Assertions.assertTrue;
public class CamelSourceTaskTest {
@@ -640,6 +642,31 @@ public class CamelSourceTaskTest {
}
}
+ @Test
+ public void testKafkaIdempotentRepositoryGetsTheClientProperties() {
+ final String clientPrefix =
CamelSourceConnectorConfig.CAMEL_CONNECTOR_IDEMPOTENCY_KAFKA_CLIENT_PREFIX;
+ final String truststore = "target/" + getClass().getSimpleName() +
"-missing-truststore.jks";
+ Map<String, String> props = new HashMap<>();
+ props.put(CamelSourceConnectorConfig.TOPIC_CONF, TOPIC_NAME);
+ props.put(CamelSourceConnectorConfig.CAMEL_SOURCE_URL_CONF,
DIRECT_URI);
+
props.put(CamelSourceConnectorConfig.CAMEL_CONNECTOR_IDEMPOTENCY_ENABLED_CONF,
"true");
+
props.put(CamelSourceConnectorConfig.CAMEL_CONNECTOR_IDEMPOTENCY_REPOSITORY_TYPE_CONF,
"kafka");
+ props.put(clientPrefix + "security.protocol", "SSL");
+ props.put(clientPrefix + "ssl.truststore.location", truststore);
+ props.put(clientPrefix + "ssl.truststore.password", "changeit");
+
+ CamelSourceTask sourceTask = new CamelSourceTask();
+ try {
+ // No broker needed: the truststore does not exist, so the
repository fails as soon as it creates its
+ // Kafka consumer ("Failed to load SSL keystore <path>"). That
only happens if the consumer was given the
+ // SSL settings under their Kafka names.
+ ConnectException e = assertThrows(ConnectException.class, () ->
sourceTask.start(props));
+ assertThat(e).hasStackTraceContaining(truststore);
+ } finally {
+ sourceTask.stop();
+ }
+ }
+
@Test
public void testSourcePollingWithIdempotencyEnabledAndHeader() {
diff --git
a/core/src/test/java/org/apache/camel/kafkaconnector/utils/IdempotentRepositoryKafkaClientPropertiesTest.java
b/core/src/test/java/org/apache/camel/kafkaconnector/utils/IdempotentRepositoryKafkaClientPropertiesTest.java
new file mode 100644
index 0000000000..adaf22ba5a
--- /dev/null
+++
b/core/src/test/java/org/apache/camel/kafkaconnector/utils/IdempotentRepositoryKafkaClientPropertiesTest.java
@@ -0,0 +1,107 @@
+/*
+ * 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.kafkaconnector.utils;
+
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Properties;
+
+import org.apache.camel.impl.DefaultCamelContext;
+import org.apache.camel.processor.idempotent.kafka.KafkaIdempotentRepository;
+import org.junit.jupiter.api.Test;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertNotSame;
+import static org.junit.jupiter.api.Assertions.assertNull;
+
+/**
+ * The Kafka idempotent repository must get the Kafka client properties of the
configuration on both its consumer and
+ * its producer, so that it can reach a cluster secured with SSL or SASL. The
repository is only built here, never
+ * started, so no broker is needed.
+ */
+public class IdempotentRepositoryKafkaClientPropertiesTest {
+
+ private static final String TOPIC = "ckc-idempotency";
+ private static final String BOOTSTRAP_SERVERS =
"broker-1:9093,broker-2:9093";
+ private static final int MAX_CACHE_SIZE = 1500;
+ private static final int POLL_DURATION_MS = 150;
+
+ private static KafkaIdempotentRepository buildRepository(Map<String, ?>
clientProperties) {
+ CamelKafkaConnectMain main =
CamelKafkaConnectMain.builder("direct://start", "log://test")
+ .withProperties(Collections.emptyMap())
+ .withIdempotencyEnabled(true)
+ .withExpressionType("body")
+ .withIdempotentRepositoryType("kafka")
+ .withIdempotentRepositoryTopicName(TOPIC)
+ .withIdempotentRepositoryKafkaServers(BOOTSTRAP_SERVERS)
+ .withIdempotentRepositoryKafkaMaxCacheSize(MAX_CACHE_SIZE)
+ .withIdempotentRepositoryKafkaPollDuration(POLL_DURATION_MS)
+ .withIdempotentRepositoryKafkaClientProperties(clientProperties)
+ .build(new DefaultCamelContext());
+
+ KafkaIdempotentRepository repository =
main.getCamelContext().getRegistry()
+ .lookupByNameAndType("ckcIdempotentRepository",
KafkaIdempotentRepository.class);
+ assertNotNull(repository, "the Kafka idempotent repository should be
bound in the registry");
+ assertEquals(TOPIC, repository.getTopic());
+ assertEquals(MAX_CACHE_SIZE, repository.getMaxCacheSize());
+ assertEquals(POLL_DURATION_MS, repository.getPollDurationMs());
+ return repository;
+ }
+
+ @Test
+ public void testClientPropertiesReachTheConsumerAndTheProducer() {
+ Map<String, String> clientProperties = new HashMap<>();
+ clientProperties.put("security.protocol", "SASL_SSL");
+ clientProperties.put("sasl.mechanism", "SCRAM-SHA-512");
+ clientProperties.put("sasl.jaas.config",
+ "org.apache.kafka.common.security.scram.ScramLoginModule
required username=\"ckc\" password=\"ckc-secret\";");
+ clientProperties.put("ssl.truststore.location",
"/opt/kafka/secrets/truststore.p12");
+ clientProperties.put("ssl.truststore.password", "truststore-secret");
+
+ KafkaIdempotentRepository repository =
buildRepository(clientProperties);
+
+ for (Properties config : List.of(repository.getConsumerConfig(),
repository.getProducerConfig())) {
+ assertEquals(BOOTSTRAP_SERVERS,
config.getProperty("bootstrap.servers"));
+ clientProperties.forEach((name, value) -> assertEquals(value,
config.getProperty(name), name));
+ assertEquals(clientProperties.size() + 1, config.size(),
"unexpected properties in " + config.keySet());
+ }
+ // the repository adds consumer-only and producer-only settings to
these when it starts
+ assertNotSame(repository.getConsumerConfig(),
repository.getProducerConfig());
+ }
+
+ @Test
+ public void
testWithoutClientPropertiesTheRepositoryIsBuiltFromTheBootstrapServersOnly() {
+ KafkaIdempotentRepository repository =
buildRepository(Collections.emptyMap());
+
+ // unchanged behaviour: the repository builds its own consumer and
producer configuration when it starts
+ assertEquals(BOOTSTRAP_SERVERS, repository.getBootstrapServers());
+ assertNull(repository.getConsumerConfig());
+ assertNull(repository.getProducerConfig());
+ }
+
+ @Test
+ public void testBootstrapServersGivenAsClientPropertyTakePrecedence() {
+ KafkaIdempotentRepository repository
+ =
buildRepository(Collections.singletonMap("bootstrap.servers",
"secured-broker:9094"));
+
+ assertEquals("secured-broker:9094",
repository.getConsumerConfig().getProperty("bootstrap.servers"));
+ assertEquals("secured-broker:9094",
repository.getProducerConfig().getProperty("bootstrap.servers"));
+ }
+}
diff --git
a/core/src/test/java/org/apache/camel/kafkaconnector/utils/StartupPropertiesLoggingTest.java
b/core/src/test/java/org/apache/camel/kafkaconnector/utils/StartupPropertiesLoggingTest.java
index b4de6f7d97..9f689feb42 100644
---
a/core/src/test/java/org/apache/camel/kafkaconnector/utils/StartupPropertiesLoggingTest.java
+++
b/core/src/test/java/org/apache/camel/kafkaconnector/utils/StartupPropertiesLoggingTest.java
@@ -16,6 +16,9 @@
*/
package org.apache.camel.kafkaconnector.utils;
+import java.io.PrintWriter;
+import java.io.StringWriter;
+import java.nio.file.Path;
import java.util.ArrayList;
import java.util.Collections;
import java.util.HashMap;
@@ -23,8 +26,11 @@ import java.util.List;
import java.util.Map;
import org.apache.camel.impl.DefaultCamelContext;
+import org.apache.camel.kafkaconnector.CamelConnectorConfig;
import org.apache.camel.kafkaconnector.CamelSinkTask;
+import org.apache.camel.kafkaconnector.CamelSourceConnectorConfig;
import org.apache.camel.kafkaconnector.CamelSourceTask;
+import org.apache.kafka.connect.errors.ConnectException;
import org.apache.logging.log4j.Level;
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.core.LogEvent;
@@ -36,8 +42,10 @@ import org.apache.logging.log4j.core.config.Property;
import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.io.TempDir;
import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertThrows;
import static org.junit.jupiter.api.Assertions.assertTrue;
/**
@@ -49,6 +57,11 @@ public class StartupPropertiesLoggingTest {
private static final String SECRET_KEY_VALUE = "sUp3rS3cr3tAccessValue";
private static final String USERINFO_PASSWORD = "hunter2PlainPassword";
+ private static final String IDEMPOTENCY_CLIENT_PREFIX =
CamelConnectorConfig.CAMEL_CONNECTOR_IDEMPOTENCY_KAFKA_CLIENT_PREFIX;
+ private static final String JAAS_PASSWORD = "jaasPl41nPassword";
+ private static final String TRUSTSTORE_PASSWORD = "trustSt0reSecret";
+ private static final String PRIVATE_KEY_BODY =
"MIIEvQIBADANBgkqhkiG9w0BAQEFAASCBKcwggSjAgEAAoIBAQC7";
+ private static final String CUSTOM_OPTION_VALUE = "cust0mL0ginSecret";
private CapturingAppender appender;
private LoggerConfig loggerConfig;
@@ -130,6 +143,72 @@ public class StartupPropertiesLoggingTest {
assertFalse(logged.contains(SECRET_KEY_VALUE), "a sensitively named
key must stay masked: " + logged);
}
+ @Test
+ public void
testIdempotentRepositoryKafkaClientSecretsAreNotLoggedInClearText() {
+ // Kafka client property names: the secret is not always in the last
segment of the key (sasl.jaas.config,
+ // ssl.keystore.key), and options of a custom login module or callback
handler can use any name at all.
+ Map<String, String> props = new HashMap<>();
+ props.put(IDEMPOTENCY_CLIENT_PREFIX + "security.protocol", "SASL_SSL");
+ props.put(IDEMPOTENCY_CLIENT_PREFIX + "sasl.jaas.config",
+ "org.apache.kafka.common.security.plain.PlainLoginModule
required username=\"ckc\" password=\"" + JAAS_PASSWORD + "\";");
+ props.put(IDEMPOTENCY_CLIENT_PREFIX + "ssl.truststore.password",
TRUSTSTORE_PASSWORD);
+ props.put(IDEMPOTENCY_CLIENT_PREFIX + "ssl.keystore.key",
+ "-----BEGIN PRIVATE KEY-----\n" + PRIVATE_KEY_BODY +
"\n-----END PRIVATE KEY-----");
+ props.put(IDEMPOTENCY_CLIENT_PREFIX + "custom.login.option",
CUSTOM_OPTION_VALUE);
+
+ String logged = startupLogLine(props);
+
+ assertFalse(logged.contains(JAAS_PASSWORD), "the JAAS configuration
must not reach the log: " + logged);
+ assertFalse(logged.contains(TRUSTSTORE_PASSWORD), "the truststore
password must not reach the log: " + logged);
+ assertFalse(logged.contains(PRIVATE_KEY_BODY), "the private key must
not reach the log: " + logged);
+ assertFalse(logged.contains(CUSTOM_OPTION_VALUE),
+ "a client property is masked whatever its name, it may be a
secret: " + logged);
+ assertTrue(logged.contains(IDEMPOTENCY_CLIENT_PREFIX +
"sasl.jaas.config"),
+ "the key should stay visible, to show the property was picked
up: " + logged);
+ }
+
+ @Test
+ public void
testIdempotentRepositoryKafkaClientSecretsDoNotLeakThroughAnyLogLine(@TempDir
Path tempDir) {
+ Map<String, String> props = new HashMap<>();
+ props.put(CamelSourceConnectorConfig.TOPIC_CONF, "my-topic");
+ props.put(CamelSourceConnectorConfig.CAMEL_SOURCE_URL_CONF,
"direct:start");
+
props.put(CamelSourceConnectorConfig.CAMEL_CONNECTOR_IDEMPOTENCY_ENABLED_CONF,
"true");
+
props.put(CamelSourceConnectorConfig.CAMEL_CONNECTOR_IDEMPOTENCY_REPOSITORY_TYPE_CONF,
"kafka");
+ props.put(IDEMPOTENCY_CLIENT_PREFIX + "security.protocol", "SASL_SSL");
+ props.put(IDEMPOTENCY_CLIENT_PREFIX + "sasl.mechanism", "PLAIN");
+ props.put(IDEMPOTENCY_CLIENT_PREFIX + "sasl.jaas.config",
+ "org.apache.kafka.common.security.plain.PlainLoginModule
required username=\"ckc\" password=\"" + JAAS_PASSWORD + "\";");
+ props.put(IDEMPOTENCY_CLIENT_PREFIX + "ssl.truststore.location",
tempDir.resolve("missing-truststore.jks").toString());
+ props.put(IDEMPOTENCY_CLIENT_PREFIX + "ssl.truststore.password",
TRUSTSTORE_PASSWORD);
+
+ CamelSourceTask task = new CamelSourceTask();
+ ConnectException failure;
+ try {
+ // the missing truststore stops the repository's Kafka consumer
right after it has logged its configuration
+ failure = assertThrows(ConnectException.class, () ->
task.start(props));
+ } finally {
+ task.stop();
+ }
+
+ List<String> logged = new ArrayList<>(appender.messages);
+ // the repository's Kafka consumer got the client properties, and
Kafka hides the secrets in its own log line
+ String consumerConfig = logged.stream()
+ .filter(m -> m.startsWith("ConsumerConfig values"))
+ .findFirst()
+ .orElseThrow(() -> new AssertionError("the repository's Kafka
consumer did not log its configuration: " + logged));
+ assertTrue(consumerConfig.contains("security.protocol = SASL_SSL"),
consumerConfig);
+ assertTrue(consumerConfig.contains("sasl.jaas.config = [hidden]"),
consumerConfig);
+ assertTrue(consumerConfig.contains("ssl.truststore.password =
[hidden]"), consumerConfig);
+ // the Kafka Connect runtime logs the failure of a task with its stack
trace
+ StringWriter failureTrace = new StringWriter();
+ failure.printStackTrace(new PrintWriter(failureTrace));
+ logged.add(failureTrace.toString());
+ for (String message : logged) {
+ assertFalse(message.contains(JAAS_PASSWORD), "the JAAS
configuration must not reach the log: " + message);
+ assertFalse(message.contains(TRUSTSTORE_PASSWORD), "the truststore
password must not reach the log: " + message);
+ }
+ }
+
@Test
public void testNonSensitiveValuesAreStillLogged() {
Map<String, String> props = new HashMap<>();
@@ -151,6 +230,11 @@ public class StartupPropertiesLoggingTest {
@Override
public void append(LogEvent event) {
messages.add(event.getMessage().getFormattedMessage());
+ if (event.getThrown() != null) {
+ StringWriter trace = new StringWriter();
+ event.getThrown().printStackTrace(new PrintWriter(trace));
+ messages.add(trace.toString());
+ }
}
}
}
diff --git a/docs/modules/ROOT/pages/user-guide/basic-configuration.adoc
b/docs/modules/ROOT/pages/user-guide/basic-configuration.adoc
index 53d2a55204..df4531cafb 100644
--- a/docs/modules/ROOT/pages/user-guide/basic-configuration.adoc
+++ b/docs/modules/ROOT/pages/user-guide/basic-configuration.adoc
@@ -28,6 +28,7 @@ For a Sink connector the basic options are:
| camel.idempotency.kafka.bootstrap.servers | A comma-separated list of host
and port pairs that are the addresses of the Kafka brokers where the idempotent
repository should live | localhost:9092 | LOW
| camel.idempotency.kafka.max.cache.size | Sets the maximum size of the local
key cache | 1000 | LOW
| camel.idempotency.kafka.poll.duration.ms | Sets the poll duration (in
milliseconds) of the Kafka consumer | 100 | LOW
+| camel.idempotency.kafka.client.<property> | A Kafka client property (for
example security.protocol, sasl.jaas.config or ssl.truststore.location) passed,
with the prefix removed, to the consumer and the producer of the Kafka
idempotent repository, see xref:user-guide/idempotency.adoc[Idempotency] | |
LOW
|===
For a Source connector the basic options are:
@@ -61,6 +62,7 @@ For a Source connector the basic options are:
| camel.idempotency.kafka.bootstrap.servers | A comma-separated list of host
and port pairs that are the addresses of the Kafka brokers where the idempotent
repository should live | localhost:9092 | LOW
| camel.idempotency.kafka.max.cache.size | Sets the maximum size of the local
key cache | 1000 | LOW
| camel.idempotency.kafka.poll.duration.ms | Sets the poll duration (in
milliseconds) of the Kafka consumer | 100 | LOW
+| camel.idempotency.kafka.client.<property> | A Kafka client property (for
example security.protocol, sasl.jaas.config or ssl.truststore.location) passed,
with the prefix removed, to the consumer and the producer of the Kafka
idempotent repository, see xref:user-guide/idempotency.adoc[Idempotency] | |
LOW
|===
For more options related to single connector you can have a look at
xref:reference/index.adoc[Connectors list].
diff --git a/docs/modules/ROOT/pages/user-guide/idempotency.adoc
b/docs/modules/ROOT/pages/user-guide/idempotency.adoc
index c464e20741..526f133bed 100644
--- a/docs/modules/ROOT/pages/user-guide/idempotency.adoc
+++ b/docs/modules/ROOT/pages/user-guide/idempotency.adoc
@@ -40,6 +40,7 @@ The idempotency feature can be enabled through a number of
configuration options
| camel.idempotency.kafka.bootstrap.servers | A comma-separated list of host
and port pairs that are the addresses of the Kafka brokers where the idempotent
repository should live | localhost:9092
| camel.idempotency.kafka.max.cache.size | Sets the maximum size of the local
key cache | 1000
| camel.idempotency.kafka.poll.duration.ms | Sets the poll duration (in
milliseconds) of the Kafka consumer | 100
+| camel.idempotency.kafka.client.<property> | A Kafka client property passed,
with the prefix removed, to the consumer and the producer of the Kafka
idempotent repository, for example
`camel.idempotency.kafka.client.security.protocol`. See
<<Idempotency-SecuredCluster,Using a secured Kafka cluster>> |
|===
The in-memory approach has been provided for short running connector workload,
while the kafka one is for long running/interruptable connector.
@@ -59,3 +60,57 @@ camel.idempotency.kafka.poll.duration.ms=150
```
Some of the options can be used with their default value, in this example
we're just listing them for a Kafka idempotent repository configuration.
+
+[[Idempotency-SecuredCluster]]
+== Using a secured Kafka cluster for the Kafka idempotent repository
+
+The Kafka idempotent repository has its own Kafka consumer and producer. They
don't use the security settings of the Kafka Connect worker, nor those of a
Kafka component configured in the connector, so with only
`camel.idempotency.kafka.bootstrap.servers` the repository can only reach an
unsecured listener.
+
+To reach a cluster secured with SSL or SASL, give the repository the Kafka
client properties it needs with the `camel.idempotency.kafka.client.` prefix.
The prefix is removed and each property is applied to both the consumer and the
producer of the repository, on top of
`camel.idempotency.kafka.bootstrap.servers` (a `bootstrap.servers` given with
the prefix takes precedence over it). Any Kafka consumer or producer property
can be passed this way, except the ones the repository always set [...]
+
+For a cluster that requires SSL, here with client authentication:
+
+[source,properties]
+----
+camel.idempotency.enabled=true
+camel.idempotency.repository.type=kafka
+camel.idempotency.kafka.topic=my.idempotency.topic
+camel.idempotency.kafka.bootstrap.servers=broker1:9093,broker2:9093
+camel.idempotency.kafka.client.security.protocol=SSL
+camel.idempotency.kafka.client.ssl.truststore.type=PKCS12
+camel.idempotency.kafka.client.ssl.truststore.location=/opt/kafka/secrets/truststore.p12
+camel.idempotency.kafka.client.ssl.truststore.password=${file:/opt/kafka/secrets/idempotency.properties:truststore.password}
+camel.idempotency.kafka.client.ssl.keystore.type=PKCS12
+camel.idempotency.kafka.client.ssl.keystore.location=/opt/kafka/secrets/keystore.p12
+camel.idempotency.kafka.client.ssl.keystore.password=${file:/opt/kafka/secrets/idempotency.properties:keystore.password}
+camel.idempotency.kafka.client.ssl.key.password=${file:/opt/kafka/secrets/idempotency.properties:key.password}
+----
+
+For a cluster that requires SASL, here SCRAM over SSL:
+
+[source,properties]
+----
+camel.idempotency.enabled=true
+camel.idempotency.repository.type=kafka
+camel.idempotency.kafka.topic=my.idempotency.topic
+camel.idempotency.kafka.bootstrap.servers=broker1:9094,broker2:9094
+camel.idempotency.kafka.client.security.protocol=SASL_SSL
+camel.idempotency.kafka.client.sasl.mechanism=SCRAM-SHA-512
+camel.idempotency.kafka.client.sasl.jaas.config=org.apache.kafka.common.security.scram.ScramLoginModule
required username="ckc"
password="${file:/opt/kafka/secrets/idempotency.properties:sasl.password}";
+camel.idempotency.kafka.client.ssl.truststore.type=PKCS12
+camel.idempotency.kafka.client.ssl.truststore.location=/opt/kafka/secrets/truststore.p12
+camel.idempotency.kafka.client.ssl.truststore.password=${file:/opt/kafka/secrets/idempotency.properties:truststore.password}
+----
+
+Rather than writing passwords and JAAS configurations in clear in the
connector configuration, keep them in a secret store and reference them, as the
`${file:...}` placeholders above do. Kafka Connect resolves these placeholders
with a config provider, declared in the worker configuration, for example the
file one:
+
+[source,properties]
+----
+config.providers=file
+config.providers.file.class=org.apache.kafka.common.config.provider.FileConfigProvider
+config.providers.file.param.allowed.paths=/opt/kafka/secrets
+----
+
+Kafka ships other config providers as well, see
https://kafka.apache.org/42/configuration/configuration-providers/[Configuration
Providers] in the Kafka documentation. The xref:security-model.adoc[security
model] describes how secrets in the connector configuration are handled.
+
+When the connector starts, the values of all the
`camel.idempotency.kafka.client.` properties are masked in the configuration it
logs, whatever their name. The consumer and the producer of the repository log
their own Kafka configuration when they are created, with the passwords and the
JAAS configuration hidden by Kafka.
