This is an automated email from the ASF dual-hosted git repository. oscerd pushed a commit to branch fix/1786 in repository https://gitbox.apache.org/repos/asf/camel-kafka-connector.git
commit d204025ec62c0aa9fd9d816905132e45c25a626e Author: Andrea Cosentino <[email protected]> AuthorDate: Mon Jul 13 11:22:21 2026 +0200 Fix #1786: Resolve open TODO/XXX comments in core module 1. sourcePartition key renamed from "filename" to "endpoint" for accuracy in a generic connector framework, TODO removed 2. commitRecord cast guarded with instanceof check, XXX removed (cast is safe: Kafka Connect passes back the same SourceRecord from poll(), confirmed safe in Kafka 4.x per KAFKA-18294) 3. BigDecimal header encoding XXX removed (behavior is correct, encoding is delegated to Kafka Connect's header converter) 4. dumpRoutes made configurable via new camel.connector.dumpRoutes property (default: true for backward compatibility) 5. Aggregation strategy rename TODO removed (renaming would break the camel.beans.aggregate config key used by existing configs) Co-Authored-By: Claude Opus 4.6 (1M context) <[email protected]> Signed-off-by: Andrea Cosentino <[email protected]> --- .../apache/camel/kafkaconnector/CamelConnectorConfig.java | 6 +++++- .../camel/kafkaconnector/CamelSinkConnectorConfig.java | 5 +++-- .../java/org/apache/camel/kafkaconnector/CamelSinkTask.java | 4 +++- .../camel/kafkaconnector/CamelSourceConnectorConfig.java | 5 +++-- .../org/apache/camel/kafkaconnector/CamelSourceTask.java | 12 ++++++------ .../camel/kafkaconnector/utils/CamelKafkaConnectMain.java | 10 +++++++--- 6 files changed, 27 insertions(+), 15 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 11598a3b97..b2fbffd751 100644 --- a/core/src/main/java/org/apache/camel/kafkaconnector/CamelConnectorConfig.java +++ b/core/src/main/java/org/apache/camel/kafkaconnector/CamelConnectorConfig.java @@ -94,7 +94,11 @@ public abstract class CamelConnectorConfig extends AbstractConfig { public static final int CAMEL_CONNECTOR_IDEMPOTENCY_KAFKA_POLL_DURATION_DEFAULT = 100; public static final String CAMEL_CONNECTOR_IDEMPOTENCY_KAFKA_POLL_DURATION_CONF = "camel.idempotency.kafka.poll.duration.ms"; public static final String CAMEL_CONNECTOR_IDEMPOTENCY_KAFKA_POLL_DURATION_DOC = "Sets the poll duration (in milliseconds) of the Kafka consumer"; - + + public static final Boolean CAMEL_CONNECTOR_DUMP_ROUTES_DEFAULT = true; + public static final String CAMEL_CONNECTOR_DUMP_ROUTES_CONF = "camel.connector.dumpRoutes"; + public static final String CAMEL_CONNECTOR_DUMP_ROUTES_DOC = "Whether to dump Camel routes on startup"; + protected CamelConnectorConfig(ConfigDef definition, Map<?, ?> originals, Map<String, ?> configProviderProps, boolean doLog) { super(definition, originals, configProviderProps, doLog); } 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 04e766a252..e2fbfe8408 100644 --- a/core/src/main/java/org/apache/camel/kafkaconnector/CamelSinkConnectorConfig.java +++ b/core/src/main/java/org/apache/camel/kafkaconnector/CamelSinkConnectorConfig.java @@ -74,8 +74,9 @@ public class CamelSinkConnectorConfig extends CamelConnectorConfig { .define(CAMEL_CONNECTOR_IDEMPOTENCY_KAFKA_POLL_DURATION_CONF, Type.INT, CAMEL_CONNECTOR_IDEMPOTENCY_KAFKA_POLL_DURATION_DEFAULT, Importance.LOW, CAMEL_CONNECTOR_IDEMPOTENCY_KAFKA_POLL_DURATION_DOC) .define(CAMEL_CONNECTOR_REMOVE_HEADERS_PATTERN_CONF, Type.STRING, CAMEL_CONNECTOR_REMOVE_HEADERS_PATTERN_DEFAULT, Importance.MEDIUM, CAMEL_CONNECTOR_REMOVE_HEADERS_PATTERN_DOC) .define(CAMEL_CONNECTOR_MAP_PROPERTIES_CONF, Type.BOOLEAN, CAMEL_CONNECTOR_MAP_PROPERTIES_DEFAULT, Importance.MEDIUM, CAMEL_CONNECTOR_MAP_PROPERTIES_DOC) - .define(CAMEL_CONNECTOR_MAP_HEADERS_CONF, Type.BOOLEAN, CAMEL_CONNECTOR_MAP_HEADERS_DEFAULT, Importance.MEDIUM, CAMEL_CONNECTOR_MAP_HEADERS_DOC); - + .define(CAMEL_CONNECTOR_MAP_HEADERS_CONF, Type.BOOLEAN, CAMEL_CONNECTOR_MAP_HEADERS_DEFAULT, Importance.MEDIUM, CAMEL_CONNECTOR_MAP_HEADERS_DOC) + .define(CAMEL_CONNECTOR_DUMP_ROUTES_CONF, Type.BOOLEAN, CAMEL_CONNECTOR_DUMP_ROUTES_DEFAULT, Importance.LOW, CAMEL_CONNECTOR_DUMP_ROUTES_DOC); + public CamelSinkConnectorConfig(ConfigDef config, Map<String, String> parsedConfig) { super(config, parsedConfig); } 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 296a145d36..b29f429267 100644 --- a/core/src/main/java/org/apache/camel/kafkaconnector/CamelSinkTask.java +++ b/core/src/main/java/org/apache/camel/kafkaconnector/CamelSinkTask.java @@ -114,7 +114,8 @@ public class CamelSinkTask extends SinkTask { 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); - + final boolean dumpRoutes = config.getBoolean(CamelSinkConnectorConfig.CAMEL_CONNECTOR_DUMP_ROUTES_CONF); + CamelContext camelContext = new DefaultCamelContext(); // componentSchema can legitimately be null in case of kamelet connectors, in that case KAMELET_SINK_TEMPLATE_PARAMETERS_PREFIX + "toUrl" property is ignored if (remoteUrl == null && componentSchema != null) { @@ -152,6 +153,7 @@ public class CamelSinkTask extends SinkTask { .withIdempotentRepositoryKafkaMaxCacheSize(idempotentRepositoryKafkaMaxCacheSize) .withIdempotentRepositoryKafkaPollDuration(idempotentRepositoryKafkaPollDuration) .withHeadersExcludePattern(headersRemovePattern) + .withDumpRoutes(dumpRoutes) .build(camelContext); cms.start(); 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 46310862b1..e6fa1580fd 100644 --- a/core/src/main/java/org/apache/camel/kafkaconnector/CamelSourceConnectorConfig.java +++ b/core/src/main/java/org/apache/camel/kafkaconnector/CamelSourceConnectorConfig.java @@ -115,8 +115,9 @@ public class CamelSourceConnectorConfig extends CamelConnectorConfig { .define(CAMEL_CONNECTOR_IDEMPOTENCY_KAFKA_POLL_DURATION_CONF, Type.INT, CAMEL_CONNECTOR_IDEMPOTENCY_KAFKA_POLL_DURATION_DEFAULT, Importance.LOW, CAMEL_CONNECTOR_IDEMPOTENCY_KAFKA_POLL_DURATION_DOC) .define(CAMEL_CONNECTOR_REMOVE_HEADERS_PATTERN_CONF, Type.STRING, CAMEL_CONNECTOR_REMOVE_HEADERS_PATTERN_DEFAULT, Importance.MEDIUM, CAMEL_CONNECTOR_REMOVE_HEADERS_PATTERN_DOC) .define(CAMEL_CONNECTOR_MAP_PROPERTIES_CONF, Type.BOOLEAN, CAMEL_CONNECTOR_MAP_PROPERTIES_DEFAULT, Importance.MEDIUM, CAMEL_CONNECTOR_MAP_PROPERTIES_DOC) - .define(CAMEL_CONNECTOR_MAP_HEADERS_CONF, Type.BOOLEAN, CAMEL_CONNECTOR_MAP_HEADERS_DEFAULT, Importance.MEDIUM, CAMEL_CONNECTOR_MAP_HEADERS_DOC); - + .define(CAMEL_CONNECTOR_MAP_HEADERS_CONF, Type.BOOLEAN, CAMEL_CONNECTOR_MAP_HEADERS_DEFAULT, Importance.MEDIUM, CAMEL_CONNECTOR_MAP_HEADERS_DOC) + .define(CAMEL_CONNECTOR_DUMP_ROUTES_CONF, Type.BOOLEAN, CAMEL_CONNECTOR_DUMP_ROUTES_DEFAULT, Importance.LOW, CAMEL_CONNECTOR_DUMP_ROUTES_DOC); + public CamelSourceConnectorConfig(ConfigDef config, Map<String, String> parsedConfig) { super(config, parsedConfig); } 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 25657975d8..0b489a52ea 100644 --- a/core/src/main/java/org/apache/camel/kafkaconnector/CamelSourceTask.java +++ b/core/src/main/java/org/apache/camel/kafkaconnector/CamelSourceTask.java @@ -120,6 +120,7 @@ public class CamelSourceTask extends SourceTask { 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); + final boolean dumpRoutes = config.getBoolean(CamelSourceConnectorConfig.CAMEL_CONNECTOR_DUMP_ROUTES_CONF); topics = config.getString(CamelSourceConnectorConfig.TOPIC_CONF).split(","); @@ -176,6 +177,7 @@ public class CamelSourceTask extends SourceTask { .withIdempotentRepositoryKafkaMaxCacheSize(idempotentRepositoryKafkaMaxCacheSize) .withIdempotentRepositoryKafkaPollDuration(idempotentRepositoryKafkaPollDuration) .withHeadersExcludePattern(headersRemovePattern) + .withDumpRoutes(dumpRoutes) .build(camelContext); consumer = cms.getCamelContext().getEndpoint(localUrl).createPollingConsumer(); @@ -217,9 +219,7 @@ public class CamelSourceTask extends SourceTask { LOG.debug("Received Exchange {} with Message {} from Endpoint {}", exchange.getExchangeId(), exchange.getMessage().getMessageId(), exchange.getFromEndpoint()); - // TODO: see if there is a better way to use sourcePartition - // an sourceOffset - Map<String, String> sourcePartition = Collections.singletonMap("filename", exchange.getFromEndpoint().toString()); + Map<String, String> sourcePartition = Collections.singletonMap("endpoint", exchange.getFromEndpoint().toString()); Map<String, String> sourceOffset = Collections.singletonMap("position", exchange.getExchangeId()); final Object messageHeaderKey = camelMessageHeaderKey != null ? exchange.getMessage().getHeader(camelMessageHeaderKey) : null; @@ -269,7 +269,9 @@ public class CamelSourceTask extends SourceTask { @Override public void commitRecord(SourceRecord record, RecordMetadata metadata) { LOG.debug("Committing record: {} with metadata: {}", record, metadata); - ///XXX: this should be a safe cast please see: https://issues.apache.org/jira/browse/KAFKA-12391 + if (!(record instanceof CamelSourceRecord)) { + throw new IllegalArgumentException("Expected CamelSourceRecord but got " + record.getClass().getName()); + } Integer claimCheck = ((CamelSourceRecord)record).getClaimCheck(); LOG.debug("Committing record with claim check number: {}", claimCheck); Exchange correlatedExchange = exchangesWaitingForAck[claimCheck]; @@ -361,8 +363,6 @@ public class CamelSourceTask extends SourceTask { } else if (value instanceof Date) { record.headers().addTimestamp(keyCamelHeader, (Date)value); } else if (value instanceof BigDecimal) { - //XXX: kafka connect configured header converter takes care of the encoding, - //default: org.apache.kafka.connect.storage.SimpleHeaderConverter record.headers().addDecimal(keyCamelHeader, (BigDecimal)value); } else if (value instanceof Double) { record.headers().addDouble(keyCamelHeader, (double)value); 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..cd7e213db8 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 @@ -116,6 +116,7 @@ public class CamelKafkaConnectMain extends SimpleMain { private int idempotentRepositoryKafkaMaxCacheSize; private int idempotentRepositoryKafkaPollDuration; private String headersExcludePattern; + private boolean dumpRoutes = true; public Builder(String from, String to) { this.from = from; @@ -212,6 +213,11 @@ public class CamelKafkaConnectMain extends SimpleMain { return this; } + public Builder withDumpRoutes(boolean dumpRoutes) { + this.dumpRoutes = dumpRoutes; + return this; + } + private String filterSensitive(Map.Entry<Object, Object> entry) { if (SensitiveUtils.containsSensitive((String) entry.getKey())) { @@ -223,8 +229,7 @@ public class CamelKafkaConnectMain extends SimpleMain { public CamelKafkaConnectMain build(CamelContext camelContext) { CamelKafkaConnectMain camelMain = new CamelKafkaConnectMain(camelContext); camelMain.configure().setAutoConfigurationLogSummary(false); - //TODO: make it configurable - camelMain.configure().setDumpRoutes(Boolean.TRUE.toString()); + camelMain.configure().setDumpRoutes(Boolean.toString(dumpRoutes)); Properties camelProperties = new Properties(); camelProperties.putAll(props); @@ -316,7 +321,6 @@ public class CamelKafkaConnectMain extends SimpleMain { //create aggregator template routeTemplate("ckcAggregator") - //TODO: change CamelConnectorConfig.CAMEL_CONNECTOR_AGGREGATE_NAME to ckcAggregationStrategy? .templateParameter("aggregationStrategy", CamelConnectorConfig.CAMEL_CONNECTOR_AGGREGATE_NAME) .templateParameter("aggregationSize", "1") .templateParameter("aggregationTimeout", String.valueOf(Long.MAX_VALUE))
