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 36e773b696 Fix #1801: reject unknown values for enum-like connector
options (#1812)
36e773b696 is described below
commit 36e773b696df3b086dd48ced06ba8c8e05425e1f
Author: Andrea Cosentino <[email protected]>
AuthorDate: Mon Aug 24 14:56:28 2026 +0200
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]>
---
.../camel/kafkaconnector/CamelConnectorConfig.java | 3 +
.../kafkaconnector/CamelSinkConnectorConfig.java | 8 +-
.../kafkaconnector/CamelSourceConnectorConfig.java | 8 +-
.../camel/kafkaconnector/ConfigValidationTest.java | 120 +++++++++++++++++++++
4 files changed, 133 insertions(+), 6 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 b2fbffd751..0bd3797762 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 e2fbfe8408..cb4cf836c2 100644
---
a/core/src/main/java/org/apache/camel/kafkaconnector/CamelSinkConnectorConfig.java
+++
b/core/src/main/java/org/apache/camel/kafkaconnector/CamelSinkConnectorConfig.java
@@ -60,14 +60,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/CamelSourceConnectorConfig.java
b/core/src/main/java/org/apache/camel/kafkaconnector/CamelSourceConnectorConfig.java
index e6fa1580fd..8132ace6fb 100644
---
a/core/src/main/java/org/apache/camel/kafkaconnector/CamelSourceConnectorConfig.java
+++
b/core/src/main/java/org/apache/camel/kafkaconnector/CamelSourceConnectorConfig.java
@@ -101,14 +101,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/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()));
+ }
+}
