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 dd3c1bc434 Fix #1786: Resolve open TODO/XXX comments in core module
(#1789)
dd3c1bc434 is described below
commit dd3c1bc4346ee771829abcf1f9727d4df1f865ee
Author: Andrea Cosentino <[email protected]>
AuthorDate: Mon Jul 13 11:49:19 2026 +0200
Fix #1786: Resolve open TODO/XXX comments in core module (#1789)
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)
Signed-off-by: Andrea Cosentino <[email protected]>
Co-authored-by: Claude Opus 4.6 (1M context) <[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))
