This is an automated email from the ASF dual-hosted git repository.
oscerd pushed a commit to branch camel-kafka-connector-4.18.x
in repository https://gitbox.apache.org/repos/asf/camel-kafka-connector.git
The following commit(s) were added to refs/heads/camel-kafka-connector-4.18.x
by this push:
new 71b6dca1e2 [backport camel-kafka-connector-4.18.x] Fixes from the
security scan triage (#1810, #1811, #1812, #1814, #1815) (#1817)
71b6dca1e2 is described below
commit 71b6dca1e290c159efa62dbf330a1af1e1b30dba
Author: Andrea Cosentino <[email protected]>
AuthorDate: Mon Aug 24 18:46:05 2026 +0200
[backport camel-kafka-connector-4.18.x] Fixes from the security scan triage
(#1810, #1811, #1812, #1814, #1815) (#1817)
* Fix #1799: FileTransforms fails the record instead of emitting a null
value (#1810)
FileTransforms.apply() carried an Eclipse-generated catch stub: an
IOException
from readFileToString was printed with e.printStackTrace() and then
execution
continued with c still null, so a file that could not be read was silently
turned into a record with a null value and a schema derived from null.
Propagate it as a ConnectException naming the file instead, so Kafka
Connect's
error handling and DLQ apply and the failure is attributed to the file that
caused it.
Also fix the else branch, which dereferenced r.value() to log the unexpected
type and therefore threw NullPointerException for a record whose value is
null,
and reuse the already-read local instead of calling r.value() three times.
Adds the first test to a connector module. The junit-jupiter test
dependency is
placed before the START OF GENERATED CODE marker in the pom so the connector
generator does not overwrite it; surefire is already configured there.
Signed-off-by: Andrea Cosentino <[email protected]>
Co-authored-by: Claude Opus 5 (1M context) <[email protected]>
(cherry picked from commit 47d918264cf786432fcf0d8861d58e35b4319c5e)
* Fix #1800: scope the SMT TypeConverter and CamelContext to the instance
(#1811)
CamelTypeConverterTransform declared its TypeConverter static but assigned
it
from the per-instance configure(), so the last SMT instance configured in a
plugin classloader repointed the converter used by every other connector's
transforms in that worker.
Make it an instance field, and release the CamelContext that
CamelTransformSupport creates per instance: close() was empty, so Kafka
Connect re-instantiating transforms on every connector config update left
those contexts accumulating for the lifetime of the worker. The core tasks
already stop their context in stop(); this brings the SMTs in line.
Also report the source and target type on a conversion failure rather than
formatting the record value into the exception message, which Kafka Connect
surfaces in the worker log and in the task status over the REST API.
The instance-scoping test registers a converter on one instance's context
only,
so it fails on a static field (the second configure() repoints it) and
passes on
an instance field - the existing tests could not tell the two apart because
each
default context resolves an equivalent converter.
Signed-off-by: Andrea Cosentino <[email protected]>
Co-authored-by: Claude Opus 5 (1M context) <[email protected]>
(cherry picked from commit d74569861497f374d86829f9d45d2159d4433dbb)
* Fix #1801: reject unknown values for enum-like connector options (#1812)
camel.idempotency.expression.type, camel.idempotency.repository.type and
camel.error.handler are each matched with a switch whose default branch does
nothing, so a misspelled value was accepted at connector submission and only
surfaced later as unrelated behaviour:
- expression.type: an unknown value left the idempotent expression unset,
so
the route template fell back to its "dummyExpression" default. simple()
on a
constant means every record after the first is treated as a duplicate.
- repository.type: an unknown value left the repository null, which was
then
bound, failing at startup with an error unrelated to the typo.
- error.handler: an unknown value silently kept a
DefaultErrorHandlerBuilder
without the configured max.redeliveries and redelivery.delay, so the
retry
policy was ignored.
Declare a ConfigDef.ValidString for each so Kafka Connect rejects the value
when
the configuration is submitted, naming the option and the allowed values.
Signed-off-by: Andrea Cosentino <[email protected]>
Co-authored-by: Claude Opus 5 (1M context) <[email protected]>
(cherry picked from commit 36e773b696df3b086dd48ced06ba8c8e05425e1f)
* Fix #1797: sanitize property values in the startup properties log (#1814)
CamelKafkaConnectMain.Builder.build() logs every initial property at INFO,
and
filterSensitive() decided what to mask from the property KEY alone. That
misses
the values that actually carry credentials:
- TaskHelper.buildUrl folds every camel.sink.endpoint.* /
camel.source.endpoint.*
option into one composed URI, stored under camel.kamelet.ckcSink.toUrl or
camel.kamelet.ckcSource.fromUrl. Neither key contains a token
SensitiveUtils
recognises, so an option that is correctly declared
ConfigDef.Type.PASSWORD
and dutifully masked under its own key was reproduced in clear text
inside
the composed URI on the same log line.
- camel.sink.url / camel.source.url may embed credentials as userinfo or as
query parameters.
Run String values through URISupport.sanitizeUri, which masks userinfo and
the
query parameters SensitiveUtils matches while leaving the endpoint itself
readable, so the log stays useful for diagnosis.
Signed-off-by: Andrea Cosentino <[email protected]>
Co-authored-by: Claude Opus 5 (1M context) <[email protected]>
(cherry picked from commit 8a310e9380b2e166764e00ed746d62f75f7ba0d5)
* Fix #1806: apply camel.remove.headers.pattern before the sink route
stages (#1815)
The route was assembled with ckcRemoveHeader appended last for both
directions:
from(from) -> [ckcMarshal] -> [ckcUnMarshal] -> [ckcAggregator]
-> [ckcIdempotent] -> ckcRemoveHeader -> toD(to)
That is correct for the source direction, where headers arrive from the
Camel
consumer and are mapped onto the produced record after the route has run.
It is
wrong for the sink direction, where CamelSinkTask.put maps
CamelHeader.-prefixed
record headers onto the exchange before it enters the route, so every stage
runs
before the removal does. With camel.remove.headers.pattern set, the
idempotency
expression, a configured AggregationStrategy and the data formats all still
saw
the headers the operator asked to strip.
Give the builder the direction through withRemoveHeadersFirst and set it in
CamelSinkTask, so the stage runs first on the sink path and stays last on
the
source path. The builder default keeps the previous placement, so a caller
driving CamelKafkaConnectMain.Builder directly is unaffected.
Signed-off-by: Andrea Cosentino <[email protected]>
Co-authored-by: Claude Opus 5 (1M context) <[email protected]>
(cherry picked from commit a859a86a0ced0e735d280271a2cbf2be1156dd29)
---------
Signed-off-by: Andrea Cosentino <[email protected]>
Co-authored-by: Claude Opus 5 (1M context) <[email protected]>
---
connectors/camel-file-kafka-connector/pom.xml | 7 +
.../file/transformers/FileTransforms.java | 15 +-
.../file/transformers/FileTransformsTest.java | 88 ++++++++++++
.../camel/kafkaconnector/CamelConnectorConfig.java | 3 +
.../kafkaconnector/CamelSinkConnectorConfig.java | 8 +-
.../apache/camel/kafkaconnector/CamelSinkTask.java | 1 +
.../kafkaconnector/CamelSourceConnectorConfig.java | 8 +-
.../transforms/CamelTransformSupport.java | 17 +++
.../transforms/CamelTypeConverterTransform.java | 6 +-
.../utils/CamelKafkaConnectMain.java | 42 +++++-
.../camel/kafkaconnector/ConfigValidationTest.java | 120 ++++++++++++++++
.../CamelTypeConverterTransformTest.java | 89 +++++++++++-
.../utils/RemoveHeadersOrderTest.java | 95 +++++++++++++
.../utils/StartupPropertiesLoggingTest.java | 156 +++++++++++++++++++++
14 files changed, 635 insertions(+), 20 deletions(-)
diff --git a/connectors/camel-file-kafka-connector/pom.xml
b/connectors/camel-file-kafka-connector/pom.xml
index 1f9170d421..c4b6d4bbf2 100644
--- a/connectors/camel-file-kafka-connector/pom.xml
+++ b/connectors/camel-file-kafka-connector/pom.xml
@@ -49,6 +49,13 @@
<artifactId>commons-io</artifactId>
<version>${commons-io-version}</version>
</dependency>
+ <!-- Test scope; version managed by junit-bom via camel-parent. Kept
outside the generated
+ block below so the connector generator does not overwrite it. -->
+ <dependency>
+ <groupId>org.junit.jupiter</groupId>
+ <artifactId>junit-jupiter</artifactId>
+ <scope>test</scope>
+ </dependency>
<!--START OF GENERATED CODE-->
<dependency>
<groupId>org.apache.camel.kafkaconnector</groupId>
diff --git
a/connectors/camel-file-kafka-connector/src/main/java/org/apache/camel/kafkaconnector/file/transformers/FileTransforms.java
b/connectors/camel-file-kafka-connector/src/main/java/org/apache/camel/kafkaconnector/file/transformers/FileTransforms.java
index 7c8244119b..cf65a78e00 100644
---
a/connectors/camel-file-kafka-connector/src/main/java/org/apache/camel/kafkaconnector/file/transformers/FileTransforms.java
+++
b/connectors/camel-file-kafka-connector/src/main/java/org/apache/camel/kafkaconnector/file/transformers/FileTransforms.java
@@ -26,6 +26,7 @@ import org.apache.camel.kafkaconnector.utils.SchemaHelper;
import org.apache.commons.io.FileUtils;
import org.apache.kafka.common.config.ConfigDef;
import org.apache.kafka.connect.connector.ConnectRecord;
+import org.apache.kafka.connect.errors.ConnectException;
import org.apache.kafka.connect.transforms.Transformation;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -40,21 +41,21 @@ public class FileTransforms<R extends ConnectRecord<R>>
implements Transformatio
public R apply(R r) {
Object value = r.value();
- if (r.value() instanceof GenericFile) {
+ if (value instanceof GenericFile) {
LOG.debug("Converting record from RemoteFile to text");
- GenericFile<File> message = (GenericFile<File>)r.value();
- String c = null;
+ GenericFile<File> message = (GenericFile<File>)value;
+ File file = message.getFile();
+ String c;
try {
- c = FileUtils.readFileToString(message.getFile(),
StandardCharsets.UTF_8);
+ c = FileUtils.readFileToString(file, StandardCharsets.UTF_8);
} catch (IOException e) {
- // TODO Auto-generated catch block
- e.printStackTrace();
+ throw new ConnectException("Failed to read file " +
file.getAbsolutePath(), e);
}
return r.newRecord(r.topic(), r.kafkaPartition(), null, r.key(),
SchemaHelper.buildSchemaBuilderForType(c), c, r.timestamp());
} else {
- LOG.debug("Unexpected message type: {}", r.value().getClass());
+ LOG.debug("Unexpected message type: {}", value == null ? null :
value.getClass());
return r;
}
diff --git
a/connectors/camel-file-kafka-connector/src/test/java/org/apache/camel/kafkaconnector/file/transformers/FileTransformsTest.java
b/connectors/camel-file-kafka-connector/src/test/java/org/apache/camel/kafkaconnector/file/transformers/FileTransformsTest.java
new file mode 100644
index 0000000000..e9f8ba6044
--- /dev/null
+++
b/connectors/camel-file-kafka-connector/src/test/java/org/apache/camel/kafkaconnector/file/transformers/FileTransformsTest.java
@@ -0,0 +1,88 @@
+/*
+ * 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.file.transformers;
+
+import java.io.File;
+import java.io.IOException;
+import java.nio.charset.StandardCharsets;
+import java.nio.file.Files;
+import java.nio.file.Path;
+
+import org.apache.camel.component.file.GenericFile;
+import org.apache.kafka.connect.errors.ConnectException;
+import org.apache.kafka.connect.source.SourceRecord;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.io.TempDir;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertInstanceOf;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertSame;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+
+public class FileTransformsTest {
+
+ @TempDir
+ private Path tempDir;
+
+ private SourceRecord recordOf(Object value) {
+ return new SourceRecord(null, null, "mytopic", 0, null, null, null,
value, null);
+ }
+
+ private SourceRecord recordFor(File file) {
+ GenericFile<File> genericFile = new GenericFile<>();
+ genericFile.setFile(file);
+ return recordOf(genericFile);
+ }
+
+ @Test
+ public void testReadsFileContentAsString() throws IOException {
+ File file = tempDir.resolve("content.txt").toFile();
+ Files.write(file.toPath(), "hello".getBytes(StandardCharsets.UTF_8));
+
+ SourceRecord transformed = new
FileTransforms<SourceRecord>().apply(recordFor(file));
+
+ assertEquals("hello", transformed.value());
+ assertNotNull(transformed.valueSchema());
+ }
+
+ @Test
+ public void testUnreadableFileFailsTheRecordInsteadOfEmittingANullValue() {
+ File missing = tempDir.resolve("missing.txt").toFile();
+
+ FileTransforms<SourceRecord> transform = new FileTransforms<>();
+ SourceRecord record = recordFor(missing);
+
+ ConnectException e = assertThrows(ConnectException.class, () ->
transform.apply(record));
+
+ assertInstanceOf(IOException.class, e.getCause());
+ }
+
+ @Test
+ public void testRecordWithANullValueIsPassedThrough() {
+ SourceRecord record = recordOf(null);
+
+ assertSame(record, new FileTransforms<SourceRecord>().apply(record));
+ }
+
+ @Test
+ public void testRecordOfAnUnexpectedTypeIsPassedThrough() {
+ SourceRecord record = recordOf("not a file");
+
+ assertSame(record, new FileTransforms<SourceRecord>().apply(record));
+ }
+}
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 11598a3b97..60b7f5b9b2 100644
---
a/core/src/main/java/org/apache/camel/kafkaconnector/CamelConnectorConfig.java
+++
b/core/src/main/java/org/apache/camel/kafkaconnector/CamelConnectorConfig.java
@@ -38,6 +38,7 @@ public abstract class CamelConnectorConfig extends
AbstractConfig {
public static final String CAMEL_CONNECTOR_ERROR_HANDLER_DEFAULT =
"default";
public static final String CAMEL_CONNECTOR_ERROR_HANDLER_CONF =
"camel.error.handler";
public static final String CAMEL_CONNECTOR_ERROR_HANDLER_DOC = "The error
handler to use: possible value are 'no' or 'default'";
+ public static final ConfigDef.Validator
CAMEL_CONNECTOR_ERROR_HANDLER_VALIDATOR = ConfigDef.ValidString.in("no",
"default");
public static final String CAMEL_CONNECTOR_REMOVE_HEADERS_PATTERN_DEFAULT
= "";
public static final String CAMEL_CONNECTOR_REMOVE_HEADERS_PATTERN_CONF =
"camel.remove.headers.pattern";
@@ -66,10 +67,12 @@ public abstract class CamelConnectorConfig extends
AbstractConfig {
public static final String
CAMEL_CONNECTOR_IDEMPOTENCY_REPOSITORY_TYPE_DEFAULT = "memory";
public static final String
CAMEL_CONNECTOR_IDEMPOTENCY_REPOSITORY_TYPE_CONF =
"camel.idempotency.repository.type";
public static final String CAMEL_CONNECTOR_IDEMPOTENCY_REPOSITORY_TYPE_DOC
= "The idempotent repository type to use, possible values are memory and kafka";
+ public static final ConfigDef.Validator
CAMEL_CONNECTOR_IDEMPOTENCY_REPOSITORY_TYPE_VALIDATOR =
ConfigDef.ValidString.in("memory", "kafka");
public static final String
CAMEL_CONNECTOR_IDEMPOTENCY_EXPRESSION_TYPE_DEFAULT = "body";
public static final String
CAMEL_CONNECTOR_IDEMPOTENCY_EXPRESSION_TYPE_CONF =
"camel.idempotency.expression.type";
public static final String CAMEL_CONNECTOR_IDEMPOTENCY_EXPRESSION_TYPE_DOC
= "How the idempotency will be evaluated: possible values are body and header";
+ public static final ConfigDef.Validator
CAMEL_CONNECTOR_IDEMPOTENCY_EXPRESSION_TYPE_VALIDATOR =
ConfigDef.ValidString.in("body", "header");
public static final String
CAMEL_CONNECTOR_IDEMPOTENCY_EXPRESSION_HEADER_DEFAULT = null;
public static final String
CAMEL_CONNECTOR_IDEMPOTENCY_EXPRESSION_HEADER_CONF =
"camel.idempotency.expression.header";
diff --git
a/core/src/main/java/org/apache/camel/kafkaconnector/CamelSinkConnectorConfig.java
b/core/src/main/java/org/apache/camel/kafkaconnector/CamelSinkConnectorConfig.java
index 1ef5a86276..11ebeb1e26 100644
---
a/core/src/main/java/org/apache/camel/kafkaconnector/CamelSinkConnectorConfig.java
+++
b/core/src/main/java/org/apache/camel/kafkaconnector/CamelSinkConnectorConfig.java
@@ -55,14 +55,16 @@ public class CamelSinkConnectorConfig extends
CamelConnectorConfig {
.define(CAMEL_CONNECTOR_AGGREGATE_CONF, Type.STRING,
CAMEL_CONNECTOR_AGGREGATE_DEFAULT, Importance.MEDIUM,
CAMEL_CONNECTOR_AGGREGATE_DOC)
.define(CAMEL_CONNECTOR_AGGREGATE_SIZE_CONF, Type.INT,
CAMEL_CONNECTOR_AGGREGATE_SIZE_DEFAULT, Importance.MEDIUM,
CAMEL_CONNECTOR_AGGREGATE_SIZE_DOC)
.define(CAMEL_CONNECTOR_AGGREGATE_TIMEOUT_CONF, Type.LONG,
CAMEL_CONNECTOR_AGGREGATE_TIMEOUT_DEFAULT, Importance.MEDIUM,
CAMEL_CONNECTOR_AGGREGATE_TIMEOUT_DOC)
- .define(CAMEL_CONNECTOR_ERROR_HANDLER_CONF, Type.STRING,
CAMEL_CONNECTOR_ERROR_HANDLER_DEFAULT, Importance.LOW,
CAMEL_CONNECTOR_ERROR_HANDLER_DOC)
+ .define(CAMEL_CONNECTOR_ERROR_HANDLER_CONF, Type.STRING,
CAMEL_CONNECTOR_ERROR_HANDLER_DEFAULT, CAMEL_CONNECTOR_ERROR_HANDLER_VALIDATOR,
Importance.LOW, CAMEL_CONNECTOR_ERROR_HANDLER_DOC)
.define(CAMEL_CONNECTOR_ERROR_HANDLER_MAXIMUM_REDELIVERIES_CONF,
Type.INT, CAMEL_CONNECTOR_ERROR_HANDLER_MAXIMUM_REDELIVERIES_DEFAULT,
Importance.MEDIUM, CAMEL_CONNECTOR_ERROR_HANDLER_MAXIMUM_REDELIVERIES_DOC)
.define(CAMEL_CONNECTOR_ERROR_HANDLER_REDELIVERY_DELAY_CONF,
Type.LONG, CAMEL_CONNECTOR_ERROR_HANDLER_REDELIVERY_DELAY_DEFAULT,
Importance.MEDIUM, CAMEL_CONNECTOR_ERROR_HANDLER_REDELIVERY_DELAY_DOC)
.define(CAMEL_CONNECTOR_IDEMPOTENCY_ENABLED_CONF, Type.BOOLEAN,
CAMEL_CONNECTOR_IDEMPOTENCY_ENABLED_DEFAULT, Importance.LOW,
CAMEL_CONNECTOR_IDEMPOTENCY_ENABLED_DOC)
- .define(CAMEL_CONNECTOR_IDEMPOTENCY_EXPRESSION_TYPE_CONF, Type.STRING,
CAMEL_CONNECTOR_IDEMPOTENCY_EXPRESSION_TYPE_DEFAULT, Importance.LOW,
CAMEL_CONNECTOR_IDEMPOTENCY_EXPRESSION_TYPE_DOC)
+ .define(CAMEL_CONNECTOR_IDEMPOTENCY_EXPRESSION_TYPE_CONF, Type.STRING,
CAMEL_CONNECTOR_IDEMPOTENCY_EXPRESSION_TYPE_DEFAULT,
+ CAMEL_CONNECTOR_IDEMPOTENCY_EXPRESSION_TYPE_VALIDATOR, Importance.LOW,
CAMEL_CONNECTOR_IDEMPOTENCY_EXPRESSION_TYPE_DOC)
.define(CAMEL_CONNECTOR_IDEMPOTENCY_EXPRESSION_HEADER_CONF,
Type.STRING, CAMEL_CONNECTOR_IDEMPOTENCY_EXPRESSION_HEADER_DEFAULT,
Importance.LOW, CAMEL_CONNECTOR_IDEMPOTENCY_EXPRESSION_HEADER_DOC)
.define(CAMEL_CONNECTOR_IDEMPOTENCY_MEMORY_DIMENSION_CONF, Type.INT,
CAMEL_CONNECTOR_IDEMPOTENCY_MEMORY_DIMENSION_DEFAULT, Importance.LOW,
CAMEL_CONNECTOR_IDEMPOTENCY_MEMORY_DIMENSION_DOC)
- .define(CAMEL_CONNECTOR_IDEMPOTENCY_REPOSITORY_TYPE_CONF, Type.STRING,
CAMEL_CONNECTOR_IDEMPOTENCY_REPOSITORY_TYPE_DEFAULT, Importance.LOW,
CAMEL_CONNECTOR_IDEMPOTENCY_REPOSITORY_TYPE_DOC)
+ .define(CAMEL_CONNECTOR_IDEMPOTENCY_REPOSITORY_TYPE_CONF, Type.STRING,
CAMEL_CONNECTOR_IDEMPOTENCY_REPOSITORY_TYPE_DEFAULT,
+ CAMEL_CONNECTOR_IDEMPOTENCY_REPOSITORY_TYPE_VALIDATOR, Importance.LOW,
CAMEL_CONNECTOR_IDEMPOTENCY_REPOSITORY_TYPE_DOC)
.define(CAMEL_CONNECTOR_IDEMPOTENCY_KAFKA_TOPIC_CONF, Type.STRING,
CAMEL_CONNECTOR_IDEMPOTENCY_KAFKA_TOPIC_DEFAULT, Importance.LOW,
CAMEL_CONNECTOR_IDEMPOTENCY_KAFKA_TOPIC_DOC)
.define(CAMEL_CONNECTOR_IDEMPOTENCY_KAFKA_BOOTSTRAP_SERVERS_CONF,
Type.STRING, CAMEL_CONNECTOR_IDEMPOTENCY_KAFKA_BOOTSTRAP_SERVERS_DEFAULT,
Importance.LOW, CAMEL_CONNECTOR_IDEMPOTENCY_KAFKA_BOOTSTRAP_SERVERS_DOC)
.define(CAMEL_CONNECTOR_IDEMPOTENCY_KAFKA_MAX_CACHE_SIZE_CONF,
Type.INT, CAMEL_CONNECTOR_IDEMPOTENCY_KAFKA_MAX_CACHE_SIZE_DEFAULT,
Importance.LOW, CAMEL_CONNECTOR_IDEMPOTENCY_KAFKA_MAX_CACHE_SIZE_DOC)
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 b66bce262e..67635d489d 100644
--- a/core/src/main/java/org/apache/camel/kafkaconnector/CamelSinkTask.java
+++ b/core/src/main/java/org/apache/camel/kafkaconnector/CamelSinkTask.java
@@ -129,6 +129,7 @@ public class CamelSinkTask extends SinkTask {
}
cms = CamelKafkaConnectMain.builder(LOCAL_URL, getSinkKamelet())
+ .withRemoveHeadersFirst(true)
.withProperties(actualProps)
.withUnmarshallDataFormat(unmarshaller)
.withMarshallDataFormat(marshaller)
diff --git
a/core/src/main/java/org/apache/camel/kafkaconnector/CamelSourceConnectorConfig.java
b/core/src/main/java/org/apache/camel/kafkaconnector/CamelSourceConnectorConfig.java
index 4acfa62a42..6de77ca0c7 100644
---
a/core/src/main/java/org/apache/camel/kafkaconnector/CamelSourceConnectorConfig.java
+++
b/core/src/main/java/org/apache/camel/kafkaconnector/CamelSourceConnectorConfig.java
@@ -96,14 +96,16 @@ public class CamelSourceConnectorConfig extends
CamelConnectorConfig {
.define(CAMEL_CONNECTOR_AGGREGATE_CONF, Type.STRING,
CAMEL_CONNECTOR_AGGREGATE_DEFAULT, Importance.MEDIUM,
CAMEL_CONNECTOR_AGGREGATE_DOC)
.define(CAMEL_CONNECTOR_AGGREGATE_SIZE_CONF, Type.INT,
CAMEL_CONNECTOR_AGGREGATE_SIZE_DEFAULT, Importance.MEDIUM,
CAMEL_CONNECTOR_AGGREGATE_SIZE_DOC)
.define(CAMEL_CONNECTOR_AGGREGATE_TIMEOUT_CONF, Type.LONG,
CAMEL_CONNECTOR_AGGREGATE_TIMEOUT_DEFAULT, Importance.MEDIUM,
CAMEL_CONNECTOR_AGGREGATE_TIMEOUT_DOC)
- .define(CAMEL_CONNECTOR_ERROR_HANDLER_CONF, Type.STRING,
CAMEL_CONNECTOR_ERROR_HANDLER_DEFAULT, Importance.LOW,
CAMEL_CONNECTOR_ERROR_HANDLER_DOC)
+ .define(CAMEL_CONNECTOR_ERROR_HANDLER_CONF, Type.STRING,
CAMEL_CONNECTOR_ERROR_HANDLER_DEFAULT, CAMEL_CONNECTOR_ERROR_HANDLER_VALIDATOR,
Importance.LOW, CAMEL_CONNECTOR_ERROR_HANDLER_DOC)
.define(CAMEL_CONNECTOR_ERROR_HANDLER_MAXIMUM_REDELIVERIES_CONF,
Type.INT, CAMEL_CONNECTOR_ERROR_HANDLER_MAXIMUM_REDELIVERIES_DEFAULT,
Importance.MEDIUM, CAMEL_CONNECTOR_ERROR_HANDLER_MAXIMUM_REDELIVERIES_DOC)
.define(CAMEL_CONNECTOR_ERROR_HANDLER_REDELIVERY_DELAY_CONF,
Type.LONG, CAMEL_CONNECTOR_ERROR_HANDLER_REDELIVERY_DELAY_DEFAULT,
Importance.MEDIUM, CAMEL_CONNECTOR_ERROR_HANDLER_REDELIVERY_DELAY_DOC)
.define(CAMEL_CONNECTOR_IDEMPOTENCY_ENABLED_CONF, Type.BOOLEAN,
CAMEL_CONNECTOR_IDEMPOTENCY_ENABLED_DEFAULT, Importance.LOW,
CAMEL_CONNECTOR_IDEMPOTENCY_ENABLED_DOC)
- .define(CAMEL_CONNECTOR_IDEMPOTENCY_EXPRESSION_TYPE_CONF, Type.STRING,
CAMEL_CONNECTOR_IDEMPOTENCY_EXPRESSION_TYPE_DEFAULT, Importance.LOW,
CAMEL_CONNECTOR_IDEMPOTENCY_EXPRESSION_TYPE_DOC)
+ .define(CAMEL_CONNECTOR_IDEMPOTENCY_EXPRESSION_TYPE_CONF, Type.STRING,
CAMEL_CONNECTOR_IDEMPOTENCY_EXPRESSION_TYPE_DEFAULT,
+ CAMEL_CONNECTOR_IDEMPOTENCY_EXPRESSION_TYPE_VALIDATOR, Importance.LOW,
CAMEL_CONNECTOR_IDEMPOTENCY_EXPRESSION_TYPE_DOC)
.define(CAMEL_CONNECTOR_IDEMPOTENCY_EXPRESSION_HEADER_CONF,
Type.STRING, CAMEL_CONNECTOR_IDEMPOTENCY_EXPRESSION_HEADER_DEFAULT,
Importance.LOW, CAMEL_CONNECTOR_IDEMPOTENCY_EXPRESSION_HEADER_DOC)
.define(CAMEL_CONNECTOR_IDEMPOTENCY_MEMORY_DIMENSION_CONF, Type.INT,
CAMEL_CONNECTOR_IDEMPOTENCY_MEMORY_DIMENSION_DEFAULT, Importance.LOW,
CAMEL_CONNECTOR_IDEMPOTENCY_MEMORY_DIMENSION_DOC)
- .define(CAMEL_CONNECTOR_IDEMPOTENCY_REPOSITORY_TYPE_CONF, Type.STRING,
CAMEL_CONNECTOR_IDEMPOTENCY_REPOSITORY_TYPE_DEFAULT, Importance.LOW,
CAMEL_CONNECTOR_IDEMPOTENCY_REPOSITORY_TYPE_DOC)
+ .define(CAMEL_CONNECTOR_IDEMPOTENCY_REPOSITORY_TYPE_CONF, Type.STRING,
CAMEL_CONNECTOR_IDEMPOTENCY_REPOSITORY_TYPE_DEFAULT,
+ CAMEL_CONNECTOR_IDEMPOTENCY_REPOSITORY_TYPE_VALIDATOR, Importance.LOW,
CAMEL_CONNECTOR_IDEMPOTENCY_REPOSITORY_TYPE_DOC)
.define(CAMEL_CONNECTOR_IDEMPOTENCY_KAFKA_TOPIC_CONF, Type.STRING,
CAMEL_CONNECTOR_IDEMPOTENCY_KAFKA_TOPIC_DEFAULT, Importance.LOW,
CAMEL_CONNECTOR_IDEMPOTENCY_KAFKA_TOPIC_DOC)
.define(CAMEL_CONNECTOR_IDEMPOTENCY_KAFKA_BOOTSTRAP_SERVERS_CONF,
Type.STRING, CAMEL_CONNECTOR_IDEMPOTENCY_KAFKA_BOOTSTRAP_SERVERS_DEFAULT,
Importance.LOW, CAMEL_CONNECTOR_IDEMPOTENCY_KAFKA_BOOTSTRAP_SERVERS_DOC)
.define(CAMEL_CONNECTOR_IDEMPOTENCY_KAFKA_MAX_CACHE_SIZE_CONF,
Type.INT, CAMEL_CONNECTOR_IDEMPOTENCY_KAFKA_MAX_CACHE_SIZE_DEFAULT,
Importance.LOW, CAMEL_CONNECTOR_IDEMPOTENCY_KAFKA_MAX_CACHE_SIZE_DOC)
diff --git
a/core/src/main/java/org/apache/camel/kafkaconnector/transforms/CamelTransformSupport.java
b/core/src/main/java/org/apache/camel/kafkaconnector/transforms/CamelTransformSupport.java
index 4f318b0ca7..64be4ace0e 100644
---
a/core/src/main/java/org/apache/camel/kafkaconnector/transforms/CamelTransformSupport.java
+++
b/core/src/main/java/org/apache/camel/kafkaconnector/transforms/CamelTransformSupport.java
@@ -20,12 +20,29 @@ import org.apache.camel.CamelContext;
import org.apache.camel.impl.DefaultCamelContext;
import org.apache.kafka.connect.connector.ConnectRecord;
import org.apache.kafka.connect.transforms.Transformation;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
public abstract class CamelTransformSupport<R extends ConnectRecord<R>>
implements Transformation<R> {
+ private static final Logger LOG =
LoggerFactory.getLogger(CamelTransformSupport.class);
+
private final CamelContext camelContext = new DefaultCamelContext();
protected CamelContext getCamelContext() {
return camelContext;
}
+
+ /**
+ * Stops the {@link CamelContext} created for this transform instance.
Kafka Connect re-instantiates transforms on
+ * every connector configuration update, so a subclass must release it
from {@link Transformation#close()} rather
+ * than let it accumulate for the lifetime of the worker.
+ */
+ protected void stopCamelContext() {
+ try {
+ camelContext.stop();
+ } catch (Exception e) {
+ LOG.warn("Failed to stop the Camel context of {}: {}",
getClass().getSimpleName(), e.getMessage(), e);
+ }
+ }
}
diff --git
a/core/src/main/java/org/apache/camel/kafkaconnector/transforms/CamelTypeConverterTransform.java
b/core/src/main/java/org/apache/camel/kafkaconnector/transforms/CamelTypeConverterTransform.java
index 0ca1162c74..2078ea21fa 100644
---
a/core/src/main/java/org/apache/camel/kafkaconnector/transforms/CamelTypeConverterTransform.java
+++
b/core/src/main/java/org/apache/camel/kafkaconnector/transforms/CamelTypeConverterTransform.java
@@ -36,7 +36,7 @@ public abstract class CamelTypeConverterTransform<R extends
ConnectRecord<R>> ex
.define(FIELD_TARGET_TYPE_CONFIG, ConfigDef.Type.CLASS, null,
ConfigDef.Importance.HIGH,
"The target field type to convert the value from, this is
full qualified Java class, e.g: java.util.Map");
- private static TypeConverter typeConverter;
+ private TypeConverter typeConverter;
private Class<?> fieldTargetType;
@Override
@@ -54,7 +54,8 @@ public abstract class CamelTypeConverterTransform<R extends
ConnectRecord<R>> ex
final Object convertedValue =
typeConverter.tryConvertTo(fieldTargetType, originalValue);
if (convertedValue == null) {
- throw new DataException(String.format("CamelTypeConverter was not
able to convert value `%s` to target type of `%s`", originalValue,
fieldTargetType.getSimpleName()));
+ throw new DataException(String.format("CamelTypeConverter was not
able to convert a value of type `%s` to target type of `%s`",
+ originalValue == null ? "null" :
originalValue.getClass().getName(), fieldTargetType.getSimpleName()));
}
return convertedValue;
@@ -80,6 +81,7 @@ public abstract class CamelTypeConverterTransform<R extends
ConnectRecord<R>> ex
@Override
public void close() {
+ stopCamelContext();
}
@Override
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 78bbeefd8b..d2c9ec4f2e 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
@@ -37,6 +37,7 @@ import
org.apache.camel.support.processor.idempotent.MemoryIdempotentRepository;
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.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -116,6 +117,7 @@ public class CamelKafkaConnectMain extends SimpleMain {
private int idempotentRepositoryKafkaMaxCacheSize;
private int idempotentRepositoryKafkaPollDuration;
private String headersExcludePattern;
+ private boolean removeHeadersFirst;
public Builder(String from, String to) {
this.from = from;
@@ -212,12 +214,39 @@ public class CamelKafkaConnectMain extends SimpleMain {
return this;
}
+ /**
+ * Controls where the header-removal stage sits in the route.
+ *
+ * On the sink path record headers are mapped onto the exchange by
{@code CamelSinkTask.put} *before* the
+ * exchange enters the route, so the stage must run first for {@code
camel.remove.headers.pattern} to keep
+ * those headers away from the marshalling, aggregation and
idempotency stages.
+ *
+ * On the source path headers arrive from the Camel consumer and are
mapped onto the produced record after
+ * the route has run, so the stage stays last and the intermediate
stages keep seeing them.
+ *
+ * @param removeHeadersFirst true for the sink direction, false (the
default) for the source direction.
+ */
+ public Builder withRemoveHeadersFirst(boolean removeHeadersFirst) {
+ this.removeHeadersFirst = removeHeadersFirst;
+ return this;
+ }
+
+
private String filterSensitive(Map.Entry<Object, Object> entry) {
+ final String key = (String) entry.getKey();
- if (SensitiveUtils.containsSensitive((String) entry.getKey())) {
- return entry.getKey() + "=xxxxxxx";
+ if (SensitiveUtils.containsSensitive(key)) {
+ return key + "=xxxxxxx";
}
- return entry.getKey() + "=" + entry.getValue();
+ final Object value = entry.getValue();
+ if (value instanceof String) {
+ // The key alone is not enough: TaskHelper.buildUrl folds
every endpoint option into a single
+ // composed URI stored under a key that carries no sensitive
token (ckcSink.toUrl /
+ // ckcSource.fromUrl), and camel.sink.url / camel.source.url
may embed credentials directly.
+ // Sanitize the value as well so userinfo and sensitive query
parameters never reach the log.
+ return key + "=" + URISupport.sanitizeUri((String) value);
+ }
+ return key + "=" + value;
}
public CamelKafkaConnectMain build(CamelContext camelContext) {
@@ -361,6 +390,9 @@ public class CamelKafkaConnectMain extends SimpleMain {
//creating the actual route
ProcessorDefinition<?> rd = from(from);
+ if (removeHeadersFirst) {
+ rd = rd.kamelet("ckcRemoveHeader");
+ }
if (!ObjectHelper.isEmpty(marshallDataFormat)) {
rd = rd.kamelet("ckcMarshal");
}
@@ -373,7 +405,9 @@ public class CamelKafkaConnectMain extends SimpleMain {
if (idempotencyEnabled) {
rd = rd.kamelet("ckcIdempotent");
}
- rd = rd.kamelet("ckcRemoveHeader");
+ if (!removeHeadersFirst) {
+ rd = rd.kamelet("ckcRemoveHeader");
+ }
rd.toD(to);
}
});
diff --git
a/core/src/test/java/org/apache/camel/kafkaconnector/ConfigValidationTest.java
b/core/src/test/java/org/apache/camel/kafkaconnector/ConfigValidationTest.java
new file mode 100644
index 0000000000..135fa8219b
--- /dev/null
+++
b/core/src/test/java/org/apache/camel/kafkaconnector/ConfigValidationTest.java
@@ -0,0 +1,120 @@
+/*
+ * 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;
+
+import java.util.HashMap;
+import java.util.Map;
+
+import org.apache.kafka.common.config.ConfigException;
+import org.junit.jupiter.api.Test;
+
+import static org.junit.jupiter.api.Assertions.assertDoesNotThrow;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+/**
+ * A behaviour-selecting option that is misspelled must be rejected when the
connector configuration is submitted,
+ * rather than silently starting a connector that behaves differently from
what the configuration asked for.
+ */
+public class ConfigValidationTest {
+
+ private Map<String, String> sinkProps() {
+ Map<String, String> props = new HashMap<>();
+ props.put("camel.sink.url", "direct://test");
+ props.put("camel.sink.kafka.topic", "mytopic");
+ return props;
+ }
+
+ private Map<String, String> sourceProps() {
+ Map<String, String> props = new HashMap<>();
+ props.put("camel.source.url", "direct://test");
+ props.put("topics", "mytopic");
+ return props;
+ }
+
+ private void assertRejected(Map<String, String> props, String option,
boolean sink) {
+ ConfigException e = assertThrows(ConfigException.class,
+ () -> {
+ if (sink) {
+ new CamelSinkConnectorConfig(props);
+ } else {
+ new CamelSourceConnectorConfig(props);
+ }
+ });
+ assertTrue(e.getMessage().contains(option), "the message should name "
+ option + " but was: " + e.getMessage());
+ }
+
+ @Test
+ public void testSinkRejectsUnknownIdempotencyExpressionType() {
+ Map<String, String> props = sinkProps();
+
props.put(CamelConnectorConfig.CAMEL_CONNECTOR_IDEMPOTENCY_EXPRESSION_TYPE_CONF,
"headers");
+ assertRejected(props,
CamelConnectorConfig.CAMEL_CONNECTOR_IDEMPOTENCY_EXPRESSION_TYPE_CONF, true);
+ }
+
+ @Test
+ public void testSinkRejectsUnknownIdempotencyRepositoryType() {
+ Map<String, String> props = sinkProps();
+
props.put(CamelConnectorConfig.CAMEL_CONNECTOR_IDEMPOTENCY_REPOSITORY_TYPE_CONF,
"in-memory");
+ assertRejected(props,
CamelConnectorConfig.CAMEL_CONNECTOR_IDEMPOTENCY_REPOSITORY_TYPE_CONF, true);
+ }
+
+ @Test
+ public void testSinkRejectsUnknownErrorHandler() {
+ Map<String, String> props = sinkProps();
+ props.put(CamelConnectorConfig.CAMEL_CONNECTOR_ERROR_HANDLER_CONF,
"none");
+ assertRejected(props,
CamelConnectorConfig.CAMEL_CONNECTOR_ERROR_HANDLER_CONF, true);
+ }
+
+ @Test
+ public void testSourceRejectsUnknownIdempotencyExpressionType() {
+ Map<String, String> props = sourceProps();
+
props.put(CamelConnectorConfig.CAMEL_CONNECTOR_IDEMPOTENCY_EXPRESSION_TYPE_CONF,
"Body");
+ assertRejected(props,
CamelConnectorConfig.CAMEL_CONNECTOR_IDEMPOTENCY_EXPRESSION_TYPE_CONF, false);
+ }
+
+ @Test
+ public void testSourceRejectsUnknownErrorHandler() {
+ Map<String, String> props = sourceProps();
+ props.put(CamelConnectorConfig.CAMEL_CONNECTOR_ERROR_HANDLER_CONF,
"retry");
+ assertRejected(props,
CamelConnectorConfig.CAMEL_CONNECTOR_ERROR_HANDLER_CONF, false);
+ }
+
+ @Test
+ public void testDocumentedValuesAreAccepted() {
+ for (String expressionType : new String[] {"body", "header"}) {
+ Map<String, String> props = sinkProps();
+
props.put(CamelConnectorConfig.CAMEL_CONNECTOR_IDEMPOTENCY_EXPRESSION_TYPE_CONF,
expressionType);
+ assertDoesNotThrow(() -> new CamelSinkConnectorConfig(props));
+ }
+ for (String repositoryType : new String[] {"memory", "kafka"}) {
+ Map<String, String> props = sinkProps();
+
props.put(CamelConnectorConfig.CAMEL_CONNECTOR_IDEMPOTENCY_REPOSITORY_TYPE_CONF,
repositoryType);
+ assertDoesNotThrow(() -> new CamelSinkConnectorConfig(props));
+ }
+ for (String errorHandler : new String[] {"no", "default"}) {
+ Map<String, String> props = sourceProps();
+ props.put(CamelConnectorConfig.CAMEL_CONNECTOR_ERROR_HANDLER_CONF,
errorHandler);
+ assertDoesNotThrow(() -> new CamelSourceConnectorConfig(props));
+ }
+ }
+
+ @Test
+ public void testDefaultsAreAccepted() {
+ assertDoesNotThrow(() -> new CamelSinkConnectorConfig(sinkProps()));
+ assertDoesNotThrow(() -> new
CamelSourceConnectorConfig(sourceProps()));
+ }
+}
diff --git
a/core/src/test/java/org/apache/camel/kafkaconnector/transforms/CamelTypeConverterTransformTest.java
b/core/src/test/java/org/apache/camel/kafkaconnector/transforms/CamelTypeConverterTransformTest.java
index 6da72c20ab..4cc93ff110 100644
---
a/core/src/test/java/org/apache/camel/kafkaconnector/transforms/CamelTypeConverterTransformTest.java
+++
b/core/src/test/java/org/apache/camel/kafkaconnector/transforms/CamelTypeConverterTransformTest.java
@@ -22,7 +22,10 @@ import java.util.HashMap;
import java.util.Map;
import io.netty.buffer.Unpooled;
+import org.apache.camel.CamelContext;
+import org.apache.camel.Exchange;
import org.apache.camel.component.netty.http.NettyChannelBufferStreamCache;
+import org.apache.camel.support.TypeConverterSupport;
import org.apache.kafka.common.config.ConfigException;
import org.apache.kafka.connect.data.Schema;
import org.apache.kafka.connect.data.SchemaBuilder;
@@ -31,8 +34,12 @@ import org.apache.kafka.connect.source.SourceRecord;
import org.apache.kafka.connect.transforms.Transformation;
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.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertInstanceOf;
import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertNotSame;
import static org.junit.jupiter.api.Assertions.assertNull;
import static org.junit.jupiter.api.Assertions.assertThrows;
import static org.junit.jupiter.api.Assertions.assertTrue;
@@ -134,4 +141,84 @@ public class CamelTypeConverterTransformTest {
assertThrows(ConfigException.class, () ->
transformationKey.configure(Collections.emptyMap()));
}
-}
\ No newline at end of file
+ @Test
+ public void testEachInstanceKeepsItsOwnCamelContext() {
+ final CamelTypeConverterTransform.Value<SourceRecord> first = new
CamelTypeConverterTransform.Value<>();
+ final CamelTypeConverterTransform.Value<SourceRecord> second = new
CamelTypeConverterTransform.Value<>();
+
+ assertNotSame(first.getCamelContext(), second.getCamelContext());
+ }
+
+ @Test
+ public void testConverterIsScopedToTheInstanceThatConfiguredIt() {
+ final CamelTypeConverterTransform.Value<SourceRecord> first = new
CamelTypeConverterTransform.Value<>();
+
+ // teach ONLY this instance's context how to produce a Marker
+
first.getCamelContext().getTypeConverterRegistry().addTypeConverter(Marker.class,
String.class,
+ new TypeConverterSupport() {
+ @Override
+ public <T> T convertTo(Class<T> type, Exchange exchange,
Object value) {
+ return type.cast(new Marker(String.valueOf(value)));
+ }
+ });
+
+ final Map<String, Object> toMarker = new HashMap<>();
+ toMarker.put(CamelTypeConverterTransform.FIELD_TARGET_TYPE_CONFIG,
Marker.class.getName());
+ first.configure(toMarker);
+
+ // configuring a second instance afterwards must not repoint the
converter the first one uses
+ final Map<String, Object> toString = new HashMap<>();
+ toString.put(CamelTypeConverterTransform.FIELD_TARGET_TYPE_CONFIG,
String.class.getName());
+ new
CamelTypeConverterTransform.Value<SourceRecord>().configure(toString);
+
+ final SourceRecord record = new SourceRecord(Collections.emptyMap(),
Collections.emptyMap(), "topic",
+ Schema.STRING_SCHEMA, "1234", Schema.STRING_SCHEMA, "abc");
+
+ assertInstanceOf(Marker.class, first.apply(record).value());
+ }
+
+ /** Target type known only to the converter registered on one instance's
context. */
+ public static final class Marker {
+ private final String value;
+
+ Marker(String value) {
+ this.value = value;
+ }
+
+ @Override
+ public String toString() {
+ return "Marker[" + value + "]";
+ }
+ }
+
+ @Test
+ public void testCloseStopsTheCamelContext() {
+ final CamelTypeConverterTransform.Value<SourceRecord> transform = new
CamelTypeConverterTransform.Value<>();
+ final Map<String, Object> props = new HashMap<>();
+ props.put(CamelTypeConverterTransform.FIELD_TARGET_TYPE_CONFIG,
String.class.getName());
+ transform.configure(props);
+
+ final CamelContext context = transform.getCamelContext();
+ assertDoesNotThrow(transform::close);
+ assertFalse(context.getStatus().isStarted());
+ }
+
+ @Test
+ public void testConversionFailureDoesNotEchoTheRecordValue() {
+ final Map<String, Object> props = new HashMap<>();
+ props.put(CamelTypeConverterTransform.FIELD_TARGET_TYPE_CONFIG,
java.time.LocalDate.class.getName());
+
+ final Transformation<SourceRecord> transform = new
CamelTypeConverterTransform.Value<>();
+ transform.configure(props);
+
+ final String secret = "s3cr3t-record-content";
+ final SourceRecord record = new SourceRecord(Collections.emptyMap(),
Collections.emptyMap(), "topic",
+ Schema.STRING_SCHEMA, "1234", Schema.STRING_SCHEMA, secret);
+
+ final org.apache.kafka.connect.errors.DataException e =
+
assertThrows(org.apache.kafka.connect.errors.DataException.class, () ->
transform.apply(record));
+
+ assertFalse(e.getMessage().contains(secret), "the record value must
not be echoed into the exception message");
+ assertTrue(e.getMessage().contains(String.class.getName()), "the
source type should be reported instead");
+ }
+}
diff --git
a/core/src/test/java/org/apache/camel/kafkaconnector/utils/RemoveHeadersOrderTest.java
b/core/src/test/java/org/apache/camel/kafkaconnector/utils/RemoveHeadersOrderTest.java
new file mode 100644
index 0000000000..0ce66ae284
--- /dev/null
+++
b/core/src/test/java/org/apache/camel/kafkaconnector/utils/RemoveHeadersOrderTest.java
@@ -0,0 +1,95 @@
+/*
+ * 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.HashMap;
+import java.util.Map;
+
+import org.apache.camel.AggregationStrategy;
+import org.apache.camel.Exchange;
+import org.apache.camel.impl.DefaultCamelContext;
+import org.apache.camel.kafkaconnector.CamelConnectorConfig;
+import org.junit.jupiter.api.Test;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertNull;
+
+/**
+ * camel.remove.headers.pattern is the documented mitigation for untrusted
headers reaching the route. On the sink
+ * path the headers are mapped onto the exchange before it enters the route,
so the removal stage has to run before
+ * the stages that consume them; on the source path they arrive from the
consumer and are read off the exchange after
+ * the route, so it stays last.
+ */
+public class RemoveHeadersOrderTest {
+
+ private static final String STRIPPED_HEADER = "CamelExecCommandExecutable";
+
+ private RecordingAggregationStrategy runRouteWith(boolean
removeHeadersFirst) throws Exception {
+ Map<String, String> props = new HashMap<>();
+
props.put(CamelConnectorConfig.CAMEL_CONNECTOR_REMOVE_HEADERS_PATTERN_CONF,
STRIPPED_HEADER);
+
+ DefaultCamelContext context = new DefaultCamelContext();
+ RecordingAggregationStrategy strategy = new
RecordingAggregationStrategy();
+
context.getRegistry().bind(CamelConnectorConfig.CAMEL_CONNECTOR_AGGREGATE_NAME,
strategy);
+
+ CamelKafkaConnectMain cms =
CamelKafkaConnectMain.builder("direct://start", "log://end")
+ .withProperties(props)
+ .withHeadersExcludePattern(STRIPPED_HEADER)
+ .withAggregationSize(1)
+ .withAggregationTimeout(1000L)
+ .withRemoveHeadersFirst(removeHeadersFirst)
+ .build(context);
+
+ cms.start();
+ try {
+ cms.getProducerTemplate().sendBodyAndHeader("direct://start",
"payload", STRIPPED_HEADER, "/bin/sh");
+ } finally {
+ cms.stop();
+ }
+ return strategy;
+ }
+
+ @Test
+ public void testSinkDirectionStripsHeadersBeforeTheAggregationStage()
throws Exception {
+ RecordingAggregationStrategy strategy = runRouteWith(true);
+
+ assertNull(strategy.seenHeaderValue,
+ "camel.remove.headers.pattern must strip the header before the
aggregation stage sees it on the "
+ + "sink path, but the strategy saw: " +
strategy.seenHeaderValue);
+ }
+
+ @Test
+ public void testSourceDirectionKeepsHeadersUntilTheEndOfTheRoute() throws
Exception {
+ RecordingAggregationStrategy strategy = runRouteWith(false);
+
+ assertEquals("/bin/sh", strategy.seenHeaderValue,
+ "on the source path the stripping stage stays last, so
intermediate stages still see the header");
+ }
+
+ private static final class RecordingAggregationStrategy implements
AggregationStrategy {
+
+ private volatile Object seenHeaderValue;
+
+ @Override
+ public Exchange aggregate(Exchange oldExchange, Exchange newExchange) {
+ if (newExchange != null) {
+ seenHeaderValue =
newExchange.getMessage().getHeader(STRIPPED_HEADER);
+ }
+ return newExchange;
+ }
+ }
+}
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
new file mode 100644
index 0000000000..b4de6f7d97
--- /dev/null
+++
b/core/src/test/java/org/apache/camel/kafkaconnector/utils/StartupPropertiesLoggingTest.java
@@ -0,0 +1,156 @@
+/*
+ * 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.ArrayList;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+import org.apache.camel.impl.DefaultCamelContext;
+import org.apache.camel.kafkaconnector.CamelSinkTask;
+import org.apache.camel.kafkaconnector.CamelSourceTask;
+import org.apache.logging.log4j.Level;
+import org.apache.logging.log4j.LogManager;
+import org.apache.logging.log4j.core.LogEvent;
+import org.apache.logging.log4j.core.LoggerContext;
+import org.apache.logging.log4j.core.appender.AbstractAppender;
+import org.apache.logging.log4j.core.config.Configuration;
+import org.apache.logging.log4j.core.config.LoggerConfig;
+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 static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+/**
+ * The connector must not write configuration secrets to the worker log at
default levels, whichever property key
+ * they reach the log under. Endpoint options are folded into a single
composed URI stored under a key that carries
+ * no sensitive token, so masking by key name alone is not enough.
+ */
+public class StartupPropertiesLoggingTest {
+
+ private static final String SECRET_KEY_VALUE = "sUp3rS3cr3tAccessValue";
+ private static final String USERINFO_PASSWORD = "hunter2PlainPassword";
+
+ private CapturingAppender appender;
+ private LoggerConfig loggerConfig;
+
+ @BeforeEach
+ public void attachAppender() {
+ appender = new CapturingAppender();
+ appender.start();
+
+ LoggerContext context = (LoggerContext) LogManager.getContext(false);
+ Configuration configuration = context.getConfiguration();
+ configuration.addAppender(appender);
+ loggerConfig =
configuration.getLoggerConfig(CamelKafkaConnectMain.class.getName());
+ loggerConfig.addAppender(appender, Level.INFO, null);
+ context.updateLoggers();
+ }
+
+ @AfterEach
+ public void detachAppender() {
+ loggerConfig.removeAppender(appender.getName());
+ appender.stop();
+ ((LoggerContext) LogManager.getContext(false)).updateLoggers();
+ }
+
+ private String startupLogLine(Map<String, String> props) {
+ CamelKafkaConnectMain.builder("direct://start", "log://test")
+ .withProperties(props)
+ .build(new DefaultCamelContext());
+
+ return appender.messages.stream()
+ .filter(m -> m.startsWith("Setting initial properties"))
+ .findFirst()
+ .orElseThrow(() -> new AssertionError("the startup properties line
was not logged: " + appender.messages));
+ }
+
+ @Test
+ public void testComposedSinkEndpointUriIsNotLoggedInClearText() {
+ Map<String, String> props = new HashMap<>();
+ props.put(CamelSinkTask.KAMELET_SINK_TEMPLATE_PARAMETERS_PREFIX +
"toUrl",
+ "aws2-kms://label?accessKey=AKIAEXAMPLEKEY&secretKey=" +
SECRET_KEY_VALUE + "®ion=eu-west-1");
+
+ String logged = startupLogLine(props);
+
+ assertFalse(logged.contains(SECRET_KEY_VALUE),
+ "the composed endpoint URI must not carry the secret into the
log: " + logged);
+ assertTrue(logged.contains("aws2-kms"), "the endpoint itself should
stay readable: " + logged);
+ }
+
+ @Test
+ public void testComposedSourceEndpointUriIsNotLoggedInClearText() {
+ Map<String, String> props = new HashMap<>();
+ props.put(CamelSourceTask.KAMELET_SOURCE_TEMPLATE_PARAMETERS_PREFIX +
"fromUrl",
+ "aws2-kms://label?accessKey=AKIAEXAMPLEKEY&secretKey=" +
SECRET_KEY_VALUE + "®ion=eu-west-1");
+
+ String logged = startupLogLine(props);
+
+ assertFalse(logged.contains(SECRET_KEY_VALUE),
+ "the composed endpoint URI must not carry the secret into the
log: " + logged);
+ }
+
+ @Test
+ public void
testCredentialsEmbeddedInTheConfiguredUrlAreNotLoggedInClearText() {
+ Map<String, String> props = new HashMap<>();
+ props.put("camel.sink.url", "netty:tcp://someuser:" +
USERINFO_PASSWORD + "@somehost:5555");
+
+ String logged = startupLogLine(props);
+
+ assertFalse(logged.contains(USERINFO_PASSWORD),
+ "userinfo credentials must not reach the log: " + logged);
+ }
+
+ @Test
+ public void testKeyBasedMaskingStillApplies() {
+ Map<String, String> props = new HashMap<>();
+ props.put("camel.sink.endpoint.secretKey", SECRET_KEY_VALUE);
+
+ String logged = startupLogLine(props);
+
+ assertFalse(logged.contains(SECRET_KEY_VALUE), "a sensitively named
key must stay masked: " + logged);
+ }
+
+ @Test
+ public void testNonSensitiveValuesAreStillLogged() {
+ Map<String, String> props = new HashMap<>();
+ props.put("camel.sink.endpoint.region", "eu-west-1");
+
+ String logged = startupLogLine(props);
+
+ assertTrue(logged.contains("eu-west-1"), "ordinary configuration
should remain visible: " + logged);
+ }
+
+ private static final class CapturingAppender extends AbstractAppender {
+
+ private final List<String> messages = Collections.synchronizedList(new
ArrayList<>());
+
+ CapturingAppender() {
+ super("ckcCapture", null, null, true, Property.EMPTY_ARRAY);
+ }
+
+ @Override
+ public void append(LogEvent event) {
+ messages.add(event.getMessage().getFormattedMessage());
+ }
+ }
+}
