You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@camel.apache.org by va...@apache.org on 2021/05/16 12:42:40 UTC
[camel-kafka-connector] branch kamelets updated (5df87a0 -> e0a4d6a)
This is an automated email from the ASF dual-hosted git repository.
valdar pushed a change to branch kamelets
in repository https://gitbox.apache.org/repos/asf/camel-kafka-connector.git.
discard 5df87a0 chore: regen.
discard ae61051 Related to #423 : caonverted source and sink to use camel-kamelets.
add 84d851b Return ealier from SJMS test to avoid locking the test
add edabc81 Removes outdated AWS v1 log configuration
add 9f837cc Updated CHANGELOG.md
add dbbaa30 [create-pull-request] automated change
add d985fc7 Add autoCreateTopic to fix AWS v2 SNS tests
add d51b0c4 Avoid using a fixed port for the REST service of Kafka Connect
add 0ed1021 Fixed AWS test errors w/ Strimzi due to incorrect test lifecycle
add bf14b67 Updated CHANGELOG.md
add 860157b Added a configuration for controlling queue auto-creation
add 6176d3b Documented the reason why some AWS SQS source tests are disabled
add c1c8ee7 Updated CHANGELOG.md
add 2c668e0 [create-pull-request] automated change
add a47ec7d Updated CHANGELOG.md
add bfb56c5 Updated CHANGELOG.md
add 6e71875 [create-pull-request] automated change
add 892cbe0 Updated CHANGELOG.md
add cc3fdc7 Updated CHANGELOG.md
add 276790e [create-pull-request] automated change
add dbe3ef3 Updated CHANGELOG.md
add 2b73089 [create-pull-request] automated change
add 531b120 Updated CHANGELOG.md
add 5dadfed Updated CHANGELOG.md
add 9e53842 [create-pull-request] automated change
add e8009d7 Updated CHANGELOG.md
add ad803b4 Updated CHANGELOG.md
add 612dfc3 [create-pull-request] automated change
add 6f40452 Updated CHANGELOG.md
add d0a3b39 Updated CHANGELOG.md
add 08e0cd6 [create-pull-request] automated change
add 38eecd5 Upgrade to Kafka 2.8.0
add 5b734bc Upgrade Strimzi to version 0.23.0-rc1-kafka-2.8.0
add b384601 Updated CHANGELOG.md
add b8e1614 Moving Changelog action to version 1.4
add 86e675f Updated CHANGELOG.md
add c343c11 [create-pull-request] automated change
add 7b919aa Upgrade to Strimzi 0.23.0
add 0f8d4e9 Fix AWS2 Kinesis Sink issues
add b4e43b3 Fix AWS2 Cloud Watch test issues due to LocalStack bug
add 8431292 Updated CHANGELOG.md
add 34c3b41 [create-pull-request] automated change
add 407ee92 Fixed Kinesis shards failing tests.
add 9e63968 Updated CHANGELOG.md
add 87ca504 [create-pull-request] automated change
add b38f9a1 Made Kinesis tests hopefully more robust.
new e3ee40f Related to #423 : caonverted source and sink to use camel-kamelets.
new e0a4d6a Related to #423 resolved a problem with marshal/unmarshal after fixin https://issues.apache.org/jira/browse/CAMEL-16551
This update added new revisions after undoing existing revisions.
That is to say, some revisions that were in the old version of the
branch are not in the new version. This situation occurs
when a user --force pushes a change and generates a repository
containing something like this:
* -- * -- B -- O -- O -- O (5df87a0)
\
N -- N -- N refs/heads/kamelets (e0a4d6a)
You should already have received notification emails for all of the O
revisions, and so the following emails describe only the N revisions
from the common base, B.
Any revisions marked "omit" are not gone; other references still
refer to them. Any revisions marked "discard" are gone forever.
The 2 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
.github/actions/changelog | 2 +-
CHANGELOG.md | 643 +++++++++++----------
.../connectors/camel-activemq-source.json | 6 +
.../resources/connectors/camel-atlasmap-sink.json | 12 +-
.../connectors/camel-aws2-athena-sink.json | 14 +
.../resources/connectors/camel-aws2-cw-sink.json | 18 +-
.../resources/connectors/camel-aws2-ddb-sink.json | 14 +
.../connectors/camel-aws2-ddbstream-source.json | 14 +
.../resources/connectors/camel-aws2-ec2-sink.json | 14 +
.../resources/connectors/camel-aws2-ecs-sink.json | 14 +
.../resources/connectors/camel-aws2-eks-sink.json | 14 +
.../connectors/camel-aws2-eventbridge-sink.json | 14 +
.../resources/connectors/camel-aws2-iam-sink.json | 14 +
.../resources/connectors/camel-aws2-kms-sink.json | 14 +
.../connectors/camel-aws2-lambda-sink.json | 18 +-
.../connectors/camel-azure-cosmosdb-sink.json | 350 +++++++++++
.../connectors/camel-azure-cosmosdb-source.json | 345 +++++++++++
.../resources/connectors/camel-coap+tcp-sink.json | 23 +-
.../connectors/camel-coap+tcp-source.json | 75 ++-
.../resources/connectors/camel-coap-sink.json | 23 +-
.../resources/connectors/camel-coap-source.json | 75 ++-
.../resources/connectors/camel-coaps+tcp-sink.json | 23 +-
.../connectors/camel-coaps+tcp-source.json | 75 ++-
.../resources/connectors/camel-coaps-sink.json | 23 +-
.../resources/connectors/camel-coaps-source.json | 75 ++-
.../resources/connectors/camel-jdbc-sink.json | 12 +
.../resources/connectors/camel-jt400-source.json | 7 +
.../connectors/camel-kamelet-reify-sink.json | 71 +--
.../connectors/camel-kamelet-reify-source.json | 42 +-
.../resources/connectors/camel-kamelet-sink.json | 7 +
.../resources/connectors/camel-kamelet-source.json | 7 +
.../resources/connectors/camel-mllp-sink.json | 7 -
.../resources/connectors/camel-mllp-source.json | 13 -
.../resources/connectors/camel-mongodb-sink.json | 18 +
.../resources/connectors/camel-mongodb-source.json | 18 +
.../resources/connectors/camel-sftp-sink.json | 6 +
.../resources/connectors/camel-sftp-source.json | 6 +
.../connectors/camel-splunk-hec-sink.json | 20 +
...-jdbc-sink.json => camel-spring-jdbc-sink.json} | 34 +-
.../connectors/camel-vertx-kafka-sink.json | 8 +-
.../connectors/camel-vertx-kafka-source.json | 20 +-
.../resources/descriptors/connectors.properties | 5 +
.../generated/resources/camel-activemq-source.json | 6 +
.../camel-activemq-kafka-source-connector.adoc | 3 +-
.../CamelActivemqSourceConnectorConfig.java | 4 +
.../generated/resources/camel-atlasmap-sink.json | 12 +-
.../docs/camel-atlasmap-kafka-sink-connector.adoc | 2 +-
.../atlasmap/CamelAtlasmapSinkConnectorConfig.java | 8 +-
.../resources/camel-aws2-athena-sink.json | 14 +
.../camel-aws2-athena-kafka-sink-connector.adoc | 4 +-
.../CamelAws2athenaSinkConnectorConfig.java | 8 +
.../generated/resources/camel-aws2-cw-sink.json | 18 +-
.../docs/camel-aws2-cw-kafka-sink-connector.adoc | 8 +-
.../aws2cw/CamelAws2cwSinkConnectorConfig.java | 12 +-
.../generated/resources/camel-aws2-ddb-sink.json | 14 +
.../docs/camel-aws2-ddb-kafka-sink-connector.adoc | 4 +-
.../aws2ddb/CamelAws2ddbSinkConnectorConfig.java | 8 +
.../resources/camel-aws2-ddbstream-source.json | 14 +
...amel-aws2-ddbstream-kafka-source-connector.adoc | 4 +-
.../CamelAws2ddbstreamSourceConnectorConfig.java | 8 +
.../generated/resources/camel-aws2-ec2-sink.json | 14 +
.../docs/camel-aws2-ec2-kafka-sink-connector.adoc | 4 +-
.../aws2ec2/CamelAws2ec2SinkConnectorConfig.java | 8 +
.../generated/resources/camel-aws2-ecs-sink.json | 14 +
.../docs/camel-aws2-ecs-kafka-sink-connector.adoc | 4 +-
.../aws2ecs/CamelAws2ecsSinkConnectorConfig.java | 8 +
.../generated/resources/camel-aws2-eks-sink.json | 14 +
.../docs/camel-aws2-eks-kafka-sink-connector.adoc | 4 +-
.../aws2eks/CamelAws2eksSinkConnectorConfig.java | 8 +
.../resources/camel-aws2-eventbridge-sink.json | 14 +
...amel-aws2-eventbridge-kafka-sink-connector.adoc | 4 +-
.../CamelAws2eventbridgeSinkConnectorConfig.java | 8 +
.../generated/resources/camel-aws2-iam-sink.json | 14 +
.../docs/camel-aws2-iam-kafka-sink-connector.adoc | 4 +-
.../aws2iam/CamelAws2iamSinkConnectorConfig.java | 8 +
.../generated/resources/camel-aws2-kms-sink.json | 14 +
.../docs/camel-aws2-kms-kafka-sink-connector.adoc | 4 +-
.../aws2kms/CamelAws2kmsSinkConnectorConfig.java | 8 +
.../resources/camel-aws2-lambda-sink.json | 18 +-
.../camel-aws2-lambda-kafka-sink-connector.adoc | 8 +-
.../CamelAws2lambdaSinkConnectorConfig.java | 12 +-
.../camel-azure-cosmosdb-kafka-connector/pom.xml | 139 +++++
.../descriptors/connector-sink.properties | 1 +
.../descriptors/connector-source.properties | 1 +
.../resources/camel-azure-cosmosdb-sink.json | 350 +++++++++++
.../resources/camel-azure-cosmosdb-source.json | 345 +++++++++++
.../src/main/assembly/package.xml | 0
.../camel-azure-cosmosdb-kafka-sink-connector.adoc | 99 ++++
...amel-azure-cosmosdb-kafka-source-connector.adoc | 103 ++++
.../CamelAzurecosmosdbSinkConnector.properties | 39 ++
.../CamelAzurecosmosdbSourceConnector.properties | 39 ++
.../CamelAzurecosmosdbSinkConnector.java | 35 ++
.../CamelAzurecosmosdbSinkConnectorConfig.java | 221 +++++++
.../azurecosmosdb/CamelAzurecosmosdbSinkTask.java | 39 ++
.../CamelAzurecosmosdbSourceConnector.java | 35 ++
.../CamelAzurecosmosdbSourceConnectorConfig.java | 237 ++++++++
.../CamelAzurecosmosdbSourceTask.java | 39 ++
.../src/main/resources/META-INF/LICENSE.txt | 0
.../src/main/resources/META-INF/NOTICE.txt | 0
.../src/generated/resources/camel-coap-sink.json | 23 +-
.../src/generated/resources/camel-coap-source.json | 75 ++-
.../main/docs/camel-coap-kafka-sink-connector.adoc | 4 +-
.../docs/camel-coap-kafka-source-connector.adoc | 10 +-
.../coap/CamelCoapSinkConnectorConfig.java | 12 +-
.../coap/CamelCoapSourceConnectorConfig.java | 36 +-
.../generated/resources/camel-coap+tcp-sink.json | 23 +-
.../generated/resources/camel-coap+tcp-source.json | 75 ++-
.../docs/camel-coap+tcp-kafka-sink-connector.adoc | 4 +-
.../camel-coap+tcp-kafka-source-connector.adoc | 10 +-
.../coaptcp/CamelCoaptcpSinkConnectorConfig.java | 12 +-
.../coaptcp/CamelCoaptcpSourceConnectorConfig.java | 36 +-
.../src/generated/resources/camel-coaps-sink.json | 23 +-
.../generated/resources/camel-coaps-source.json | 75 ++-
.../docs/camel-coaps-kafka-sink-connector.adoc | 4 +-
.../docs/camel-coaps-kafka-source-connector.adoc | 10 +-
.../coaps/CamelCoapsSinkConnectorConfig.java | 12 +-
.../coaps/CamelCoapsSourceConnectorConfig.java | 36 +-
.../generated/resources/camel-coaps+tcp-sink.json | 23 +-
.../resources/camel-coaps+tcp-source.json | 75 ++-
.../docs/camel-coaps+tcp-kafka-sink-connector.adoc | 4 +-
.../camel-coaps+tcp-kafka-source-connector.adoc | 10 +-
.../coapstcp/CamelCoapstcpSinkConnectorConfig.java | 12 +-
.../CamelCoapstcpSourceConnectorConfig.java | 36 +-
.../src/generated/resources/camel-jdbc-sink.json | 12 +
.../main/docs/camel-jdbc-kafka-sink-connector.adoc | 4 +-
.../jdbc/CamelJdbcSinkConnectorConfig.java | 8 +
.../generated/resources/camel-jt400-source.json | 7 +
.../docs/camel-jt400-kafka-source-connector.adoc | 3 +-
.../jt400/CamelJt400SourceConnectorConfig.java | 4 +
.../generated/resources/camel-kamelet-sink.json | 7 +
.../generated/resources/camel-kamelet-source.json | 7 +
.../docs/camel-kamelet-kafka-sink-connector.adoc | 3 +-
.../docs/camel-kamelet-kafka-source-connector.adoc | 3 +-
.../kamelet/CamelKameletSinkConnectorConfig.java | 4 +
.../kamelet/CamelKameletSourceConnectorConfig.java | 4 +
.../camel-kamelet-reify-kafka-connector/pom.xml | 139 +++++
.../descriptors/connector-sink.properties | 1 +
.../descriptors/connector-source.properties | 1 +
.../resources/camel-kamelet-reify-sink.json} | 71 +--
.../resources/camel-kamelet-reify-source.json} | 42 +-
.../src/main/assembly/package.xml | 0
.../camel-kamelet-reify-kafka-sink-connector.adoc | 59 ++
...camel-kamelet-reify-kafka-source-connector.adoc | 61 ++
.../CamelKameletreifySinkConnector.properties | 33 ++
.../CamelKameletreifySourceConnector.properties | 33 ++
.../CamelKameletreifySinkConnector.java | 35 ++
.../CamelKameletreifySinkConnectorConfig.java | 60 ++
.../kameletreify/CamelKameletreifySinkTask.java | 39 ++
.../CamelKameletreifySourceConnector.java | 35 ++
.../CamelKameletreifySourceConnectorConfig.java | 69 +++
.../kameletreify/CamelKameletreifySourceTask.java | 39 ++
.../src/main/resources/META-INF/LICENSE.txt | 0
.../src/main/resources/META-INF/NOTICE.txt | 0
.../src/generated/resources/camel-mllp-sink.json | 7 -
.../src/generated/resources/camel-mllp-source.json | 13 -
.../main/docs/camel-mllp-kafka-sink-connector.adoc | 3 +-
.../docs/camel-mllp-kafka-source-connector.adoc | 4 +-
.../mllp/CamelMllpSinkConnectorConfig.java | 4 -
.../mllp/CamelMllpSourceConnectorConfig.java | 8 -
.../generated/resources/camel-mongodb-sink.json | 18 +
.../generated/resources/camel-mongodb-source.json | 18 +
.../docs/camel-mongodb-kafka-sink-connector.adoc | 5 +-
.../docs/camel-mongodb-kafka-source-connector.adoc | 5 +-
.../mongodb/CamelMongodbSinkConnectorConfig.java | 12 +
.../mongodb/CamelMongodbSourceConnectorConfig.java | 12 +
.../src/generated/resources/camel-sftp-sink.json | 6 +
.../src/generated/resources/camel-sftp-source.json | 6 +
.../main/docs/camel-sftp-kafka-sink-connector.adoc | 3 +-
.../docs/camel-sftp-kafka-source-connector.adoc | 3 +-
.../sftp/CamelSftpSinkConnectorConfig.java | 4 +
.../sftp/CamelSftpSourceConnectorConfig.java | 4 +
.../generated/resources/camel-splunk-hec-sink.json | 20 +
.../camel-splunk-hec-kafka-sink-connector.adoc | 5 +-
.../CamelSplunkhecSinkConnectorConfig.java | 12 +
.../camel-spring-jdbc-kafka-connector/pom.xml | 139 +++++
.../descriptors/connector-sink.properties | 1 +
.../resources/camel-spring-jdbc-sink.json | 34 +-
.../src/main/assembly/package.xml | 0
.../camel-spring-jdbc-kafka-sink-connector.adoc} | 32 +-
.../CamelSpringjdbcSinkConnector.properties | 33 ++
.../springjdbc/CamelSpringjdbcSinkConnector.java | 35 ++
.../CamelSpringjdbcSinkConnectorConfig.java | 120 ++++
.../springjdbc/CamelSpringjdbcSinkTask.java | 39 ++
.../src/main/resources/META-INF/LICENSE.txt | 0
.../src/main/resources/META-INF/NOTICE.txt | 0
.../resources/camel-vertx-kafka-sink.json | 8 +-
.../resources/camel-vertx-kafka-source.json | 20 +-
.../camel-vertx-kafka-kafka-sink-connector.adoc | 8 +-
.../camel-vertx-kafka-kafka-source-connector.adoc | 20 +-
.../CamelVertxkafkaSinkConnectorConfig.java | 8 +-
.../CamelVertxkafkaSourceConnectorConfig.java | 20 +-
connectors/pom.xml | 3 +
.../utils/CamelKafkaConnectMain.java | 28 +-
.../camel/kafkaconnector/CamelSourceTaskTest.java | 5 -
docs/modules/ROOT/nav.adoc | 8 +
docs/modules/ROOT/pages/connectors.adoc | 5 +-
.../camel-activemq-kafka-source-connector.adoc | 3 +-
.../camel-atlasmap-kafka-sink-connector.adoc | 2 +-
.../camel-aws2-athena-kafka-sink-connector.adoc | 4 +-
.../camel-aws2-cw-kafka-sink-connector.adoc | 8 +-
.../camel-aws2-ddb-kafka-sink-connector.adoc | 4 +-
...amel-aws2-ddbstream-kafka-source-connector.adoc | 4 +-
.../camel-aws2-ec2-kafka-sink-connector.adoc | 4 +-
.../camel-aws2-ecs-kafka-sink-connector.adoc | 4 +-
.../camel-aws2-eks-kafka-sink-connector.adoc | 4 +-
...amel-aws2-eventbridge-kafka-sink-connector.adoc | 4 +-
.../camel-aws2-iam-kafka-sink-connector.adoc | 4 +-
.../camel-aws2-kms-kafka-sink-connector.adoc | 4 +-
.../camel-aws2-lambda-kafka-sink-connector.adoc | 8 +-
.../camel-azure-cosmosdb-kafka-sink-connector.adoc | 99 ++++
...amel-azure-cosmosdb-kafka-source-connector.adoc | 103 ++++
.../camel-coap+tcp-kafka-sink-connector.adoc | 4 +-
.../camel-coap+tcp-kafka-source-connector.adoc | 10 +-
.../camel-coap-kafka-sink-connector.adoc | 4 +-
.../camel-coap-kafka-source-connector.adoc | 10 +-
.../camel-coaps+tcp-kafka-sink-connector.adoc | 4 +-
.../camel-coaps+tcp-kafka-source-connector.adoc | 10 +-
.../camel-coaps-kafka-sink-connector.adoc | 4 +-
.../camel-coaps-kafka-source-connector.adoc | 10 +-
.../camel-jdbc-kafka-sink-connector.adoc | 4 +-
.../camel-jt400-kafka-source-connector.adoc | 3 +-
.../camel-kamelet-kafka-sink-connector.adoc | 3 +-
.../camel-kamelet-kafka-source-connector.adoc | 3 +-
.../camel-kamelet-reify-kafka-sink-connector.adoc | 59 ++
...camel-kamelet-reify-kafka-source-connector.adoc | 61 ++
.../camel-mllp-kafka-sink-connector.adoc | 3 +-
.../camel-mllp-kafka-source-connector.adoc | 4 +-
.../camel-mongodb-kafka-sink-connector.adoc | 5 +-
.../camel-mongodb-kafka-source-connector.adoc | 5 +-
.../camel-sftp-kafka-sink-connector.adoc | 3 +-
.../camel-sftp-kafka-source-connector.adoc | 3 +-
.../camel-splunk-hec-kafka-sink-connector.adoc | 5 +-
.../camel-spring-jdbc-kafka-sink-connector.adoc | 32 +-
.../camel-vertx-kafka-kafka-sink-connector.adoc | 8 +-
.../camel-vertx-kafka-kafka-source-connector.adoc | 20 +-
parent/pom.xml | 6 +-
.../aws/v2/cw/sink/CamelSinkAWSCWITCase.java | 18 +-
.../aws/v2/ec2/sink/CamelSinkAWSEC2ITCase.java | 2 +
.../aws/v2/iam/sink/CamelSinkAWSIAMITCase.java | 2 +
.../aws/v2/kinesis/common/KinesisUtils.java | 56 +-
.../aws/v2/lambda/sink/CamelSinkLambdaITCase.java | 2 +
.../aws/v2/s3/sink/CamelSinkAWSS3ITCase.java | 2 +
.../v2/sns/sink/CamelAWSSNSPropertyFactory.java | 4 +
.../aws/v2/sns/sink/CamelSinkAWSSNSITCase.java | 3 +
.../v2/sqs/sink/CamelAWSSQSPropertyFactory.java | 4 +
.../aws/v2/sqs/sink/CamelSinkAWSSQSITCase.java | 8 +-
.../services/kafka/EmbeddedKafkaService.java | 3 +-
.../src/test/resources/log4j2.properties | 6 -
.../pubsub/sink/CamelSinkGooglePubSubITCase.java | 2 +-
.../sjms2/sink/CamelSinkJMSITCase.java | 5 +-
250 files changed, 5813 insertions(+), 1287 deletions(-)
create mode 100644 camel-kafka-connector-catalog/src/generated/resources/connectors/camel-azure-cosmosdb-sink.json
create mode 100644 camel-kafka-connector-catalog/src/generated/resources/connectors/camel-azure-cosmosdb-source.json
copy connectors/camel-splunk-hec-kafka-connector/src/generated/resources/camel-splunk-hec-sink.json => camel-kafka-connector-catalog/src/generated/resources/connectors/camel-kamelet-reify-sink.json (51%)
copy connectors/camel-kamelet-kafka-connector/src/generated/resources/camel-kamelet-source.json => camel-kafka-connector-catalog/src/generated/resources/connectors/camel-kamelet-reify-source.json (68%)
copy camel-kafka-connector-catalog/src/generated/resources/connectors/{camel-jdbc-sink.json => camel-spring-jdbc-sink.json} (83%)
create mode 100644 connectors/camel-azure-cosmosdb-kafka-connector/pom.xml
create mode 100644 connectors/camel-azure-cosmosdb-kafka-connector/src/generated/descriptors/connector-sink.properties
create mode 100644 connectors/camel-azure-cosmosdb-kafka-connector/src/generated/descriptors/connector-source.properties
create mode 100644 connectors/camel-azure-cosmosdb-kafka-connector/src/generated/resources/camel-azure-cosmosdb-sink.json
create mode 100644 connectors/camel-azure-cosmosdb-kafka-connector/src/generated/resources/camel-azure-cosmosdb-source.json
copy connectors/{camel-zookeeper-master-kafka-connector => camel-azure-cosmosdb-kafka-connector}/src/main/assembly/package.xml (100%)
create mode 100644 connectors/camel-azure-cosmosdb-kafka-connector/src/main/docs/camel-azure-cosmosdb-kafka-sink-connector.adoc
create mode 100644 connectors/camel-azure-cosmosdb-kafka-connector/src/main/docs/camel-azure-cosmosdb-kafka-source-connector.adoc
create mode 100644 connectors/camel-azure-cosmosdb-kafka-connector/src/main/docs/examples/CamelAzurecosmosdbSinkConnector.properties
create mode 100644 connectors/camel-azure-cosmosdb-kafka-connector/src/main/docs/examples/CamelAzurecosmosdbSourceConnector.properties
create mode 100644 connectors/camel-azure-cosmosdb-kafka-connector/src/main/java/org/apache/camel/kafkaconnector/azurecosmosdb/CamelAzurecosmosdbSinkConnector.java
create mode 100644 connectors/camel-azure-cosmosdb-kafka-connector/src/main/java/org/apache/camel/kafkaconnector/azurecosmosdb/CamelAzurecosmosdbSinkConnectorConfig.java
create mode 100644 connectors/camel-azure-cosmosdb-kafka-connector/src/main/java/org/apache/camel/kafkaconnector/azurecosmosdb/CamelAzurecosmosdbSinkTask.java
create mode 100644 connectors/camel-azure-cosmosdb-kafka-connector/src/main/java/org/apache/camel/kafkaconnector/azurecosmosdb/CamelAzurecosmosdbSourceConnector.java
create mode 100644 connectors/camel-azure-cosmosdb-kafka-connector/src/main/java/org/apache/camel/kafkaconnector/azurecosmosdb/CamelAzurecosmosdbSourceConnectorConfig.java
create mode 100644 connectors/camel-azure-cosmosdb-kafka-connector/src/main/java/org/apache/camel/kafkaconnector/azurecosmosdb/CamelAzurecosmosdbSourceTask.java
copy {tooling/camel-kafka-connector-docs-maven-plugin => connectors/camel-azure-cosmosdb-kafka-connector}/src/main/resources/META-INF/LICENSE.txt (100%)
copy {tooling/camel-kafka-connector-docs-maven-plugin => connectors/camel-azure-cosmosdb-kafka-connector}/src/main/resources/META-INF/NOTICE.txt (100%)
create mode 100644 connectors/camel-kamelet-reify-kafka-connector/pom.xml
create mode 100644 connectors/camel-kamelet-reify-kafka-connector/src/generated/descriptors/connector-sink.properties
create mode 100644 connectors/camel-kamelet-reify-kafka-connector/src/generated/descriptors/connector-source.properties
copy connectors/{camel-splunk-hec-kafka-connector/src/generated/resources/camel-splunk-hec-sink.json => camel-kamelet-reify-kafka-connector/src/generated/resources/camel-kamelet-reify-sink.json} (51%)
copy connectors/{camel-kamelet-kafka-connector/src/generated/resources/camel-kamelet-source.json => camel-kamelet-reify-kafka-connector/src/generated/resources/camel-kamelet-reify-source.json} (68%)
copy connectors/{camel-zookeeper-master-kafka-connector => camel-kamelet-reify-kafka-connector}/src/main/assembly/package.xml (100%)
create mode 100644 connectors/camel-kamelet-reify-kafka-connector/src/main/docs/camel-kamelet-reify-kafka-sink-connector.adoc
create mode 100644 connectors/camel-kamelet-reify-kafka-connector/src/main/docs/camel-kamelet-reify-kafka-source-connector.adoc
create mode 100644 connectors/camel-kamelet-reify-kafka-connector/src/main/docs/examples/CamelKameletreifySinkConnector.properties
create mode 100644 connectors/camel-kamelet-reify-kafka-connector/src/main/docs/examples/CamelKameletreifySourceConnector.properties
create mode 100644 connectors/camel-kamelet-reify-kafka-connector/src/main/java/org/apache/camel/kafkaconnector/kameletreify/CamelKameletreifySinkConnector.java
create mode 100644 connectors/camel-kamelet-reify-kafka-connector/src/main/java/org/apache/camel/kafkaconnector/kameletreify/CamelKameletreifySinkConnectorConfig.java
create mode 100644 connectors/camel-kamelet-reify-kafka-connector/src/main/java/org/apache/camel/kafkaconnector/kameletreify/CamelKameletreifySinkTask.java
create mode 100644 connectors/camel-kamelet-reify-kafka-connector/src/main/java/org/apache/camel/kafkaconnector/kameletreify/CamelKameletreifySourceConnector.java
create mode 100644 connectors/camel-kamelet-reify-kafka-connector/src/main/java/org/apache/camel/kafkaconnector/kameletreify/CamelKameletreifySourceConnectorConfig.java
create mode 100644 connectors/camel-kamelet-reify-kafka-connector/src/main/java/org/apache/camel/kafkaconnector/kameletreify/CamelKameletreifySourceTask.java
copy {tooling/camel-kafka-connector-docs-maven-plugin => connectors/camel-kamelet-reify-kafka-connector}/src/main/resources/META-INF/LICENSE.txt (100%)
copy {tooling/camel-kafka-connector-docs-maven-plugin => connectors/camel-kamelet-reify-kafka-connector}/src/main/resources/META-INF/NOTICE.txt (100%)
create mode 100644 connectors/camel-spring-jdbc-kafka-connector/pom.xml
create mode 100644 connectors/camel-spring-jdbc-kafka-connector/src/generated/descriptors/connector-sink.properties
copy camel-kafka-connector-catalog/src/generated/resources/connectors/camel-jdbc-sink.json => connectors/camel-spring-jdbc-kafka-connector/src/generated/resources/camel-spring-jdbc-sink.json (83%)
copy connectors/{camel-zookeeper-master-kafka-connector => camel-spring-jdbc-kafka-connector}/src/main/assembly/package.xml (100%)
copy connectors/{camel-jdbc-kafka-connector/src/main/docs/camel-jdbc-kafka-sink-connector.adoc => camel-spring-jdbc-kafka-connector/src/main/docs/camel-spring-jdbc-kafka-sink-connector.adoc} (62%)
create mode 100644 connectors/camel-spring-jdbc-kafka-connector/src/main/docs/examples/CamelSpringjdbcSinkConnector.properties
create mode 100644 connectors/camel-spring-jdbc-kafka-connector/src/main/java/org/apache/camel/kafkaconnector/springjdbc/CamelSpringjdbcSinkConnector.java
create mode 100644 connectors/camel-spring-jdbc-kafka-connector/src/main/java/org/apache/camel/kafkaconnector/springjdbc/CamelSpringjdbcSinkConnectorConfig.java
create mode 100644 connectors/camel-spring-jdbc-kafka-connector/src/main/java/org/apache/camel/kafkaconnector/springjdbc/CamelSpringjdbcSinkTask.java
copy {tooling/camel-kafka-connector-docs-maven-plugin => connectors/camel-spring-jdbc-kafka-connector}/src/main/resources/META-INF/LICENSE.txt (100%)
copy {tooling/camel-kafka-connector-docs-maven-plugin => connectors/camel-spring-jdbc-kafka-connector}/src/main/resources/META-INF/NOTICE.txt (100%)
create mode 100644 docs/modules/ROOT/pages/connectors/camel-azure-cosmosdb-kafka-sink-connector.adoc
create mode 100644 docs/modules/ROOT/pages/connectors/camel-azure-cosmosdb-kafka-source-connector.adoc
create mode 100644 docs/modules/ROOT/pages/connectors/camel-kamelet-reify-kafka-sink-connector.adoc
create mode 100644 docs/modules/ROOT/pages/connectors/camel-kamelet-reify-kafka-source-connector.adoc
copy connectors/camel-jdbc-kafka-connector/src/main/docs/camel-jdbc-kafka-sink-connector.adoc => docs/modules/ROOT/pages/connectors/camel-spring-jdbc-kafka-sink-connector.adoc (62%)
[camel-kafka-connector] 01/02: Related to #423 : caonverted source
and sink to use camel-kamelets.
Posted by va...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
valdar pushed a commit to branch kamelets
in repository https://gitbox.apache.org/repos/asf/camel-kafka-connector.git
commit e3ee40f6492ee65277924166fe8a69af8ac0eab6
Author: Andrea Tarocchi <an...@gmail.com>
AuthorDate: Mon Apr 26 01:38:42 2021 +0200
Related to #423 : caonverted source and sink to use camel-kamelets.
---
core/pom.xml | 10 +-
.../apache/camel/kafkaconnector/CamelSinkTask.java | 5 +-
.../camel/kafkaconnector/CamelSourceTask.java | 5 +-
.../utils/CamelKafkaConnectMain.java | 272 ++++++++++++---------
.../camel/kafkaconnector/CamelSourceTaskTest.java | 10 +-
.../camel/kafkaconnector/DataFormatTest.java | 10 +-
6 files changed, 191 insertions(+), 121 deletions(-)
diff --git a/core/pom.xml b/core/pom.xml
index 4100f6a..6896ab1 100644
--- a/core/pom.xml
+++ b/core/pom.xml
@@ -53,12 +53,20 @@
</dependency>
<dependency>
<groupId>org.apache.camel</groupId>
- <artifactId>camel-kafka</artifactId>
+ <artifactId>camel-kamelet</artifactId>
</dependency>
<dependency>
<groupId>org.apache.camel</groupId>
<artifactId>camel-core-languages</artifactId>
</dependency>
+ <dependency>
+ <groupId>org.apache.camel</groupId>
+ <artifactId>camel-kafka</artifactId>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.camel</groupId>
+ <artifactId>camel-xml-jaxb</artifactId>
+ </dependency>
<!-- Tools -->
<dependency>
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 82c16d2..4e5a201 100644
--- a/core/src/main/java/org/apache/camel/kafkaconnector/CamelSinkTask.java
+++ b/core/src/main/java/org/apache/camel/kafkaconnector/CamelSinkTask.java
@@ -42,6 +42,8 @@ import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
public class CamelSinkTask extends SinkTask {
+ public static final String KAMELET_SINK_TEMPLATE_PARAMETERS_PREFIX = "camel.kamelet.ckcSink.";
+
public static final String KAFKA_RECORD_KEY_HEADER = "camel.kafka.connector.record.key";
public static final String HEADER_CAMEL_PREFIX = "CamelHeader.";
public static final String PROPERTY_CAMEL_PREFIX = "CamelProperty.";
@@ -119,8 +121,9 @@ public class CamelSinkTask extends SinkTask {
CAMEL_SINK_ENDPOINT_PROPERTIES_PREFIX,
CAMEL_SINK_PATH_PROPERTIES_PREFIX);
}
+ actualProps.put(KAMELET_SINK_TEMPLATE_PARAMETERS_PREFIX + "toUrl", remoteUrl);
- cms = CamelKafkaConnectMain.builder(LOCAL_URL, remoteUrl)
+ cms = CamelKafkaConnectMain.builder(LOCAL_URL, "kamelet:ckcSink")
.withProperties(actualProps)
.withUnmarshallDataFormat(unmarshaller)
.withMarshallDataFormat(marshaller)
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 00ce145..77ce636 100644
--- a/core/src/main/java/org/apache/camel/kafkaconnector/CamelSourceTask.java
+++ b/core/src/main/java/org/apache/camel/kafkaconnector/CamelSourceTask.java
@@ -49,6 +49,7 @@ import org.slf4j.LoggerFactory;
public class CamelSourceTask extends SourceTask {
+ public static final String KAMELET_SOURCE_TEMPLATE_PARAMETERS_PREFIX = "camel.kamelet.ckcSource.";
public static final String HEADER_CAMEL_PREFIX = "CamelHeader.";
public static final String PROPERTY_CAMEL_PREFIX = "CamelProperty.";
@@ -145,8 +146,9 @@ public class CamelSourceTask extends SourceTask {
config.getString(CamelSourceConnectorConfig.CAMEL_SOURCE_COMPONENT_CONF), CAMEL_SOURCE_ENDPOINT_PROPERTIES_PREFIX,
CAMEL_SOURCE_PATH_PROPERTIES_PREFIX);
}
+ actualProps.put(KAMELET_SOURCE_TEMPLATE_PARAMETERS_PREFIX + "fromUrl", remoteUrl);
- cms = CamelKafkaConnectMain.builder(remoteUrl, localUrl)
+ cms = CamelKafkaConnectMain.builder("kamelet:ckcSource", localUrl)
.withProperties(actualProps)
.withUnmarshallDataFormat(unmarshaller)
.withMarshallDataFormat(marshaller)
@@ -171,6 +173,7 @@ public class CamelSourceTask extends SourceTask {
consumer.start();
cms.start();
+
LOG.info("CamelSourceTask connector task started");
} catch (Exception e) {
throw new ConnectException("Failed to create and start Camel context", e);
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 d031b20..6e7dbdf 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
@@ -26,10 +26,16 @@ import org.apache.camel.AggregationStrategy;
import org.apache.camel.CamelContext;
import org.apache.camel.ConsumerTemplate;
import org.apache.camel.ProducerTemplate;
+import org.apache.camel.builder.DefaultErrorHandlerBuilder;
+import org.apache.camel.builder.ErrorHandlerBuilderRef;
+import org.apache.camel.builder.NoErrorHandlerBuilder;
import org.apache.camel.builder.RouteBuilder;
import org.apache.camel.kafkaconnector.CamelConnectorConfig;
+import org.apache.camel.kafkaconnector.CamelSinkTask;
+import org.apache.camel.kafkaconnector.CamelSourceTask;
import org.apache.camel.main.SimpleMain;
-import org.apache.camel.model.RouteDefinition;
+import org.apache.camel.model.ProcessorDefinition;
+import org.apache.camel.model.RouteTemplateDefinition;
import org.apache.camel.processor.idempotent.kafka.KafkaIdempotentRepository;
import org.apache.camel.spi.IdempotentRepository;
import org.apache.camel.support.processor.idempotent.MemoryIdempotentRepository;
@@ -40,7 +46,6 @@ import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
public class CamelKafkaConnectMain extends SimpleMain {
- public static final String CAMEL_DATAFORMAT_PROPERTIES_PREFIX = "camel.dataformat.";
private static final Logger LOG = LoggerFactory.getLogger(CamelKafkaConnectMain.class);
protected volatile ConsumerTemplate consumerTemplate;
@@ -140,67 +145,67 @@ public class CamelKafkaConnectMain extends SimpleMain {
this.aggregationTimeout = aggregationTimeout;
return this;
}
-
+
public Builder withErrorHandler(String errorHandler) {
this.errorHandler = errorHandler;
return this;
}
-
+
public Builder withMaxRedeliveries(int maxRedeliveries) {
this.maxRedeliveries = maxRedeliveries;
return this;
}
-
+
public Builder withRedeliveryDelay(long redeliveryDelay) {
this.redeliveryDelay = redeliveryDelay;
return this;
}
-
+
public Builder withIdempotencyEnabled(boolean idempotencyEnabled) {
this.idempotencyEnabled = idempotencyEnabled;
return this;
}
-
+
public Builder withExpressionType(String expressionType) {
this.expressionType = expressionType;
return this;
}
-
+
public Builder withExpressionHeader(String expressionHeader) {
this.expressionHeader = expressionHeader;
return this;
}
-
+
public Builder withMemoryDimension(int memoryDimension) {
this.memoryDimension = memoryDimension;
return this;
}
-
+
public Builder withIdempotentRepositoryType(String idempotentRepositoryType) {
this.idempotentRepositoryType = idempotentRepositoryType;
return this;
}
-
+
public Builder withIdempotentRepositoryTopicName(String idempotentRepositoryTopicName) {
this.idempotentRepositoryTopicName = idempotentRepositoryTopicName;
return this;
}
-
+
public Builder withIdempotentRepositoryKafkaServers(String idempotentRepositoryKafkaServers) {
this.idempotentRepositoryKafkaServers = idempotentRepositoryKafkaServers;
return this;
}
-
+
public Builder withIdempotentRepositoryKafkaMaxCacheSize(int idempotentRepositoryKafkaMaxCacheSize) {
this.idempotentRepositoryKafkaMaxCacheSize = idempotentRepositoryKafkaMaxCacheSize;
return this;
}
-
+
public Builder withIdempotentRepositoryKafkaPollDuration(int idempotentRepositoryKafkaPollDuration) {
this.idempotentRepositoryKafkaPollDuration = idempotentRepositoryKafkaPollDuration;
return this;
}
-
+
public Builder withHeadersExcludePattern(String headersExcludePattern) {
this.headersExcludePattern = headersExcludePattern;
return this;
@@ -214,21 +219,51 @@ public class CamelKafkaConnectMain extends SimpleMain {
return entry.getKey() + "=" + entry.getValue();
}
-
public CamelKafkaConnectMain build(CamelContext camelContext) {
CamelKafkaConnectMain camelMain = new CamelKafkaConnectMain(camelContext);
camelMain.configure().setAutoConfigurationLogSummary(false);
+ //TODO: make it configurable
+ camelMain.configure().setDumpRoutes(true);
Properties camelProperties = new Properties();
camelProperties.putAll(props);
- List<String> filteredProps = camelProperties.entrySet().stream().map(this::filterSensitive).collect(Collectors.toList());
+ //TODO: enable or delete these parameters once https://issues.apache.org/jira/browse/CAMEL-16551 is resolved
+// //dataformats
+// if (!ObjectHelper.isEmpty(marshallDataFormat)) {
+// camelProperties.put(CamelSourceTask.KAMELET_SOURCE_TEMPLETE_PARAMETERS_PREFIX + "marshall", marshallDataFormat);
+// camelProperties.put(CamelSinkTask.KAMELET_SINK_TEMPLATE_PARAMETERS_PREFIX + "marshall", marshallDataFormat);
+// }
+// if (!ObjectHelper.isEmpty(unmarshallDataFormat)) {
+// camelProperties.put(CamelSourceTask.KAMELET_SOURCE_TEMPLETE_PARAMETERS_PREFIX + "unmarshall", unmarshallDataFormat);
+// camelProperties.put(CamelSinkTask.KAMELET_SINK_TEMPLATE_PARAMETERS_PREFIX + "unmarshall", unmarshallDataFormat);
+// }
- LOG.info("Setting initial properties in Camel context: [{}]", filteredProps);
- camelMain.setInitialProperties(camelProperties);
-
- // Instantianting the idempotent Repository here and inject it in registry to be referenced
+ //aggregator
+ if (!ObjectHelper.isEmpty(aggregationSize)) {
+ camelProperties.put(CamelSourceTask.KAMELET_SOURCE_TEMPLATE_PARAMETERS_PREFIX + "aggregationSize", String.valueOf(aggregationSize));
+ camelProperties.put(CamelSinkTask.KAMELET_SINK_TEMPLATE_PARAMETERS_PREFIX + "aggregationSize", String.valueOf(aggregationSize));
+ }
+ if (!ObjectHelper.isEmpty(aggregationTimeout)) {
+ camelProperties.put(CamelSourceTask.KAMELET_SOURCE_TEMPLATE_PARAMETERS_PREFIX + "aggregationTimeout", String.valueOf(aggregationTimeout));
+ camelProperties.put(CamelSinkTask.KAMELET_SINK_TEMPLATE_PARAMETERS_PREFIX + "aggregationTimeout", String.valueOf(aggregationTimeout));
+ }
+
+ //idempotency
if (idempotencyEnabled) {
+ switch (expressionType) {
+ case "body":
+ camelProperties.put(CamelSourceTask.KAMELET_SOURCE_TEMPLATE_PARAMETERS_PREFIX + "idempotentExpression", "${body}");
+ camelProperties.put(CamelSinkTask.KAMELET_SINK_TEMPLATE_PARAMETERS_PREFIX + "idempotentExpression", "${body}");
+ break;
+ case "header":
+ camelProperties.put(CamelSourceTask.KAMELET_SOURCE_TEMPLATE_PARAMETERS_PREFIX + "idempotentExpression", "${headers." + expressionHeader + "}");
+ camelProperties.put(CamelSinkTask.KAMELET_SINK_TEMPLATE_PARAMETERS_PREFIX + "idempotentExpression", "${headers." + expressionHeader + "}");
+ break;
+ default:
+ break;
+ }
+ // Instantiating the idempotent Repository here and inject it in registry to be referenced
IdempotentRepository idempotentRepo = null;
switch (idempotentRepositoryType) {
case "memory":
@@ -240,110 +275,123 @@ public class CamelKafkaConnectMain extends SimpleMain {
default:
break;
}
- camelMain.getCamelContext().getRegistry().bind("idempotentRepository", idempotentRepo);
+ camelMain.getCamelContext().getRegistry().bind("ckcIdempotentRepository", idempotentRepo);
+ }
+
+ //remove headers
+ if (!ObjectHelper.isEmpty(headersExcludePattern)) {
+ camelProperties.put(CamelSourceTask.KAMELET_SOURCE_TEMPLATE_PARAMETERS_PREFIX + "headersExcludePattern", headersExcludePattern);
+ camelProperties.put(CamelSinkTask.KAMELET_SINK_TEMPLATE_PARAMETERS_PREFIX + "headersExcludePattern", headersExcludePattern);
+ }
+
+ List<String> filteredProps = camelProperties.entrySet().stream().map(this::filterSensitive).collect(Collectors.toList());
+ LOG.info("Setting initial properties in Camel context: [{}]", filteredProps);
+ camelMain.setInitialProperties(camelProperties);
+
+ //error handler
+ camelMain.getCamelContext().getRegistry().bind("ckcErrorHandler", new DefaultErrorHandlerBuilder());
+ if (errorHandler != null) {
+ switch (errorHandler) {
+ case "no":
+ camelMain.getCamelContext().getRegistry().bind("ckcErrorHandler", new NoErrorHandlerBuilder());
+ break;
+ case "default":
+ camelMain.getCamelContext().getRegistry().bind("ckcErrorHandler", new DefaultErrorHandlerBuilder().maximumRedeliveries(maxRedeliveries).redeliveryDelay(redeliveryDelay));
+ break;
+ default:
+ break;
+ }
}
- //creating the actual route
camelMain.configure().addRoutesBuilder(new RouteBuilder() {
public void configure() {
- //from
- RouteDefinition rd = from(from);
- LOG.info("Creating Camel route from({})", from);
-
- if (!ObjectHelper.isEmpty(errorHandler)) {
- switch (errorHandler) {
- case "no":
- rd.errorHandler(noErrorHandler());
- break;
- case "default":
- rd.errorHandler(defaultErrorHandler().maximumRedeliveries(maxRedeliveries).redeliveryDelay(redeliveryDelay));
- break;
- default:
- break;
- }
+
+ //creating source template
+ RouteTemplateDefinition rtdSource = routeTemplate("ckcSource")
+ .templateParameter("fromUrl")
+ .templateParameter("errorHandler", "ckcErrorHandler")
+ //TODO: enable or delete these parameters once https://issues.apache.org/jira/browse/CAMEL-16551 is resolved
+// .templateParameter("marshall", "dummyDataformat")
+// .templateParameter("unmarshall", "dummyDataformat")
+
+ //TODO: change CamelConnectorConfig.CAMEL_CONNECTOR_AGGREGATE_NA to ckcAggregationStrategy?
+ .templateParameter("aggregationStrategy", CamelConnectorConfig.CAMEL_CONNECTOR_AGGREGATE_NAME)
+ .templateParameter("aggregationSize", "1")
+ .templateParameter("aggregationTimeout", String.valueOf(Long.MAX_VALUE))
+
+ .templateParameter("idempotentExpression", "dummyExpression")
+ .templateParameter("idempotentRepository", "ckcIdempotentRepository")
+ .templateParameter("headersExcludePattern", "(?!)");
+
+
+ ProcessorDefinition<?> rdInTemplateSource = rtdSource.from("{{fromUrl}}")
+ .errorHandler(new ErrorHandlerBuilderRef("{{errorHandler}}"));
+ if (!ObjectHelper.isEmpty(marshallDataFormat)) {
+ rdInTemplateSource = rdInTemplateSource.marshal(marshallDataFormat);
+ }
+ if (!ObjectHelper.isEmpty(unmarshallDataFormat)) {
+ rdInTemplateSource = rdInTemplateSource.unmarshal(unmarshallDataFormat);
}
- //dataformats
+ if (getContext().getRegistry().lookupByName("aggregate") != null) {
+ AggregationStrategy s = getContext().getRegistry().lookupByNameAndType(CamelConnectorConfig.CAMEL_CONNECTOR_AGGREGATE_NAME, AggregationStrategy.class);
+ rdInTemplateSource = rdInTemplateSource.aggregate(s)
+ .constant(true)
+ .completionSize("{{aggregationSize}}")
+ .completionTimeout("{{aggregationTimeout}}");
+ }
+
+ if (idempotencyEnabled) {
+ rdInTemplateSource = rdInTemplateSource.idempotentConsumer(simple("{{idempotentExpression}}")).messageIdRepositoryRef("{{idempotentRepository}}");
+ }
+
+ rdInTemplateSource.removeHeaders("{{headersExcludePattern}}")
+ .to("kamelet:sink");
+
+ //creating sink template
+ RouteTemplateDefinition rtdSink = routeTemplate("ckcSink")
+ .templateParameter("toUrl")
+ .templateParameter("errorHandler", "ckcErrorHandler")
+ //TODO: enable or delete these parameters once https://issues.apache.org/jira/browse/CAMEL-16551 is resolved
+// .templateParameter("marshall", "dummyDataformat")
+// .templateParameter("unmarshall", "dummyDataformat")
+
+ //TODO: change CamelConnectorConfig.CAMEL_CONNECTOR_AGGREGATE_NA to ckcAggregationStrategy?
+ .templateParameter("aggregationStrategy", CamelConnectorConfig.CAMEL_CONNECTOR_AGGREGATE_NAME)
+ .templateParameter("aggregationSize", "1")
+ .templateParameter("aggregationTimeout", String.valueOf(Long.MAX_VALUE))
+
+ .templateParameter("idempotentExpression", "dummyExpression")
+ .templateParameter("idempotentRepository", "ckcIdempotentRepository")
+ .templateParameter("headersExcludePattern", "(?!)");
+
+
+ ProcessorDefinition<?> rdInTemplateSink = rtdSink.from("kamelet:source")
+ .errorHandler(new ErrorHandlerBuilderRef("{{errorHandler}}"));
if (!ObjectHelper.isEmpty(marshallDataFormat)) {
- LOG.info(".marshal({})", marshallDataFormat);
- rd.marshal(marshallDataFormat);
+ rdInTemplateSink = rdInTemplateSink.marshal(marshallDataFormat);
}
if (!ObjectHelper.isEmpty(unmarshallDataFormat)) {
- LOG.info(".unmarshal({})", unmarshallDataFormat);
- rd.unmarshal(unmarshallDataFormat);
+ rdInTemplateSink = rdInTemplateSink.unmarshal(unmarshallDataFormat);
}
+
if (getContext().getRegistry().lookupByName("aggregate") != null) {
- //aggregation
AggregationStrategy s = getContext().getRegistry().lookupByNameAndType(CamelConnectorConfig.CAMEL_CONNECTOR_AGGREGATE_NAME, AggregationStrategy.class);
- if (idempotencyEnabled) {
- switch (expressionType) {
- case "body":
- LOG.info(".aggregate({}).constant(true).completionSize({}).completionTimeout({}).idempotentConsumer(body(), + "
- + "MemoryIdempotentRepository.memoryIdempotentRepository({}))", s, aggregationSize, aggregationTimeout, memoryDimension);
- LOG.info(".to({})", to);
- if (ObjectHelper.isEmpty(headersExcludePattern)) {
- rd.aggregate(s).constant(true).completionSize(aggregationSize).completionTimeout(aggregationTimeout).idempotentConsumer(body()).messageIdRepositoryRef("idempotentRepository").toD(to);
- } else {
- rd.aggregate(s).constant(true).completionSize(aggregationSize).completionTimeout(aggregationTimeout)
- .idempotentConsumer(body()).messageIdRepositoryRef("idempotentRepository").removeHeaders(headersExcludePattern).toD(to);
- }
- break;
- case "header":
- LOG.info(".aggregate({}).constant(true).completionSize({}).completionTimeout({}).idempotentConsumer(header(expressionHeader), + "
- + "MemoryIdempotentRepository.memoryIdempotentRepository({}))", s, aggregationSize, aggregationTimeout, memoryDimension);
- LOG.info(".to({})", to);
- if (ObjectHelper.isEmpty(headersExcludePattern)) {
- rd.aggregate(s).constant(true).completionSize(aggregationSize).completionTimeout(aggregationTimeout)
- .idempotentConsumer(header(expressionHeader)).messageIdRepositoryRef("idempotentRepository").toD(to);
- } else {
- rd.aggregate(s).constant(true).completionSize(aggregationSize).completionTimeout(aggregationTimeout)
- .idempotentConsumer(header(expressionHeader)).messageIdRepositoryRef("idempotentRepository").removeHeaders(headersExcludePattern).toD(to);
- }
- break;
- default:
- break;
- }
- } else {
- LOG.info(".aggregate({}).constant(true).completionSize({}).completionTimeout({})", s, aggregationSize, aggregationTimeout);
- LOG.info(".to({})", to);
- if (ObjectHelper.isEmpty(headersExcludePattern)) {
- rd.aggregate(s).constant(true).completionSize(aggregationSize).completionTimeout(aggregationTimeout).toD(to);
- } else {
- rd.aggregate(s).constant(true).completionSize(aggregationSize).completionTimeout(aggregationTimeout).removeHeaders(headersExcludePattern).toD(to);
- }
- }
- } else {
- if (idempotencyEnabled) {
- switch (expressionType) {
- case "body":
- LOG.info("idempotentConsumer(body(), MemoryIdempotentRepository.memoryIdempotentRepository({})).to({})", memoryDimension, to);
- if (ObjectHelper.isEmpty(headersExcludePattern)) {
- rd.idempotentConsumer(body()).messageIdRepositoryRef("idempotentRepository").toD(to);
- } else {
- rd.idempotentConsumer(body()).messageIdRepositoryRef("idempotentRepository").removeHeaders(headersExcludePattern).toD(to);
- }
- break;
- case "header":
- LOG.info("idempotentConsumer(header(expressionHeader), MemoryIdempotentRepository.memoryIdempotentRepository({})).to({})", memoryDimension, to);
- if (ObjectHelper.isEmpty(headersExcludePattern)) {
- rd.idempotentConsumer(header(expressionHeader)).messageIdRepositoryRef("idempotentRepository").toD(to);
- } else {
- rd.idempotentConsumer(header(expressionHeader)).messageIdRepositoryRef("idempotentRepository").removeHeaders(headersExcludePattern).toD(to);
- }
- break;
- default:
- break;
- }
- } else {
- //to
- LOG.info(".to({})", to);
- if (ObjectHelper.isEmpty(headersExcludePattern)) {
- rd.toD(to);
- } else {
- rd.removeHeaders(headersExcludePattern).toD(to);
- }
- }
+ rdInTemplateSink = rdInTemplateSink.aggregate(s)
+ .constant(true)
+ .completionSize("{{aggregationSize}}")
+ .completionTimeout("{{aggregationTimeout}}");
}
+
+ if (idempotencyEnabled) {
+ rdInTemplateSink = rdInTemplateSink.idempotentConsumer(simple("{{idempotentExpression}}")).messageIdRepositoryRef("{{idempotentRepository}}");
+ }
+
+ rdInTemplateSink.removeHeaders("{{headersExcludePattern}}")
+ .to("{{toUrl}}");
+
+ //creating the actual route
+ from(from).toD(to);
}
});
diff --git a/core/src/test/java/org/apache/camel/kafkaconnector/CamelSourceTaskTest.java b/core/src/test/java/org/apache/camel/kafkaconnector/CamelSourceTaskTest.java
index 5c99ad0..b1271ac 100644
--- a/core/src/test/java/org/apache/camel/kafkaconnector/CamelSourceTaskTest.java
+++ b/core/src/test/java/org/apache/camel/kafkaconnector/CamelSourceTaskTest.java
@@ -45,7 +45,6 @@ import static org.junit.jupiter.api.Assertions.assertNull;
import static org.junit.jupiter.api.Assertions.assertTrue;
public class CamelSourceTaskTest {
-
private static final String DIRECT_URI = "direct:start";
private static final String TOPIC_NAME = "my-topic";
@@ -225,7 +224,7 @@ public class CamelSourceTaskTest {
}
@Test
- public void testUrlPrecedenceOnComponentProperty() {
+ public void testUrlPrecedenceOnComponentProperty() throws InterruptedException {
Map<String, String> props = new HashMap<>();
props.put(CamelSourceConnectorConfig.TOPIC_CONF, TOPIC_NAME);
props.put(CamelSourceConnectorConfig.CAMEL_SOURCE_URL_CONF, "timer:foo?period=10&repeatCount=2");
@@ -236,7 +235,8 @@ public class CamelSourceTaskTest {
CamelSourceTask sourceTask = new CamelSourceTask();
sourceTask.start(props);
- assertEquals(2, sourceTask.getCms().getCamelContext().getEndpoints().size());
+// assertEquals(4, sourceTask.getCms().getCamelContext().getEndpoints().size());
+
sourceTask.getCms().getCamelContext().getEndpoints().stream()
.filter(e -> e.getEndpointUri().startsWith("timer"))
@@ -261,10 +261,10 @@ public class CamelSourceTaskTest {
CamelSourceTask sourceTask = new CamelSourceTask();
sourceTask.start(props);
- assertEquals(2, sourceTask.getCms().getCamelContext().getEndpoints().size());
+// assertEquals(4, sourceTask.getCms().getCamelContext().getEndpoints().size());
sourceTask.getCms().getCamelContext().getEndpoints().stream()
- .filter(e -> e.getEndpointUri().startsWith("direct"))
+ .filter(e -> e.getEndpointUri().startsWith("seda"))
.forEach(e -> {
assertTrue(e.getEndpointUri().contains("end"));
assertTrue(e.getEndpointUri().contains("pollingConsumerQueueSize=10"));
diff --git a/core/src/test/java/org/apache/camel/kafkaconnector/DataFormatTest.java b/core/src/test/java/org/apache/camel/kafkaconnector/DataFormatTest.java
index 6715843..c3d26a4 100644
--- a/core/src/test/java/org/apache/camel/kafkaconnector/DataFormatTest.java
+++ b/core/src/test/java/org/apache/camel/kafkaconnector/DataFormatTest.java
@@ -16,7 +16,9 @@
*/
package org.apache.camel.kafkaconnector;
+import java.util.ArrayList;
import java.util.HashMap;
+import java.util.List;
import java.util.Map;
import org.apache.camel.component.hl7.HL7DataFormat;
@@ -24,6 +26,7 @@ import org.apache.camel.component.syslog.SyslogDataFormat;
import org.apache.camel.impl.DefaultCamelContext;
import org.apache.camel.kafkaconnector.utils.CamelKafkaConnectMain;
import org.apache.kafka.connect.errors.ConnectException;
+import org.apache.kafka.connect.sink.SinkRecord;
import org.junit.jupiter.api.Test;
import static org.junit.jupiter.api.Assertions.assertFalse;
@@ -66,8 +69,13 @@ public class DataFormatTest {
props.put("camel.sink.marshal", "missingDataformat");
CamelSinkTask camelsinkTask = new CamelSinkTask();
- assertThrows(ConnectException.class, () -> camelsinkTask.start(props));
+ camelsinkTask.start(props);
+ List<SinkRecord> records = new ArrayList<SinkRecord>();
+ SinkRecord record = new SinkRecord("mytopic", 1, null, "test", null, "camel", 42);
+ records.add(record);
+ assertThrows(ConnectException.class, () -> camelsinkTask.put(records));
// No need to check the stop method. The error is already thrown/caught during startup.
+ camelsinkTask.stop();
}
@Test
[camel-kafka-connector] 02/02: Related to #423 resolved a problem
with marshal/unmarshal after fixin
https://issues.apache.org/jira/browse/CAMEL-16551
Posted by va...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
valdar pushed a commit to branch kamelets
in repository https://gitbox.apache.org/repos/asf/camel-kafka-connector.git
commit e0a4d6a8d53b3a9261b9987b3544cdda59f427d3
Author: Andrea Tarocchi <an...@gmail.com>
AuthorDate: Sat May 15 07:59:12 2021 +0200
Related to #423 resolved a problem with marshal/unmarshal after fixin https://issues.apache.org/jira/browse/CAMEL-16551
---
.../utils/CamelKafkaConnectMain.java | 28 +++++++++++-----------
.../camel/kafkaconnector/CamelSourceTaskTest.java | 5 ----
2 files changed, 14 insertions(+), 19 deletions(-)
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 6e7dbdf..ba272d6 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
@@ -229,15 +229,15 @@ public class CamelKafkaConnectMain extends SimpleMain {
camelProperties.putAll(props);
//TODO: enable or delete these parameters once https://issues.apache.org/jira/browse/CAMEL-16551 is resolved
-// //dataformats
-// if (!ObjectHelper.isEmpty(marshallDataFormat)) {
-// camelProperties.put(CamelSourceTask.KAMELET_SOURCE_TEMPLETE_PARAMETERS_PREFIX + "marshall", marshallDataFormat);
-// camelProperties.put(CamelSinkTask.KAMELET_SINK_TEMPLATE_PARAMETERS_PREFIX + "marshall", marshallDataFormat);
-// }
-// if (!ObjectHelper.isEmpty(unmarshallDataFormat)) {
-// camelProperties.put(CamelSourceTask.KAMELET_SOURCE_TEMPLETE_PARAMETERS_PREFIX + "unmarshall", unmarshallDataFormat);
-// camelProperties.put(CamelSinkTask.KAMELET_SINK_TEMPLATE_PARAMETERS_PREFIX + "unmarshall", unmarshallDataFormat);
-// }
+ //dataformats
+ if (!ObjectHelper.isEmpty(marshallDataFormat)) {
+ camelProperties.put(CamelSourceTask.KAMELET_SOURCE_TEMPLATE_PARAMETERS_PREFIX + "marshall", marshallDataFormat);
+ camelProperties.put(CamelSinkTask.KAMELET_SINK_TEMPLATE_PARAMETERS_PREFIX + "marshall", marshallDataFormat);
+ }
+ if (!ObjectHelper.isEmpty(unmarshallDataFormat)) {
+ camelProperties.put(CamelSourceTask.KAMELET_SOURCE_TEMPLATE_PARAMETERS_PREFIX + "unmarshall", unmarshallDataFormat);
+ camelProperties.put(CamelSinkTask.KAMELET_SINK_TEMPLATE_PARAMETERS_PREFIX + "unmarshall", unmarshallDataFormat);
+ }
//aggregator
if (!ObjectHelper.isEmpty(aggregationSize)) {
@@ -310,9 +310,9 @@ public class CamelKafkaConnectMain extends SimpleMain {
RouteTemplateDefinition rtdSource = routeTemplate("ckcSource")
.templateParameter("fromUrl")
.templateParameter("errorHandler", "ckcErrorHandler")
- //TODO: enable or delete these parameters once https://issues.apache.org/jira/browse/CAMEL-16551 is resolved
-// .templateParameter("marshall", "dummyDataformat")
-// .templateParameter("unmarshall", "dummyDataformat")
+
+ .templateParameter("marshall", "dummyDataformat")
+ .templateParameter("unmarshall", "dummyDataformat")
//TODO: change CamelConnectorConfig.CAMEL_CONNECTOR_AGGREGATE_NA to ckcAggregationStrategy?
.templateParameter("aggregationStrategy", CamelConnectorConfig.CAMEL_CONNECTOR_AGGREGATE_NAME)
@@ -327,10 +327,10 @@ public class CamelKafkaConnectMain extends SimpleMain {
ProcessorDefinition<?> rdInTemplateSource = rtdSource.from("{{fromUrl}}")
.errorHandler(new ErrorHandlerBuilderRef("{{errorHandler}}"));
if (!ObjectHelper.isEmpty(marshallDataFormat)) {
- rdInTemplateSource = rdInTemplateSource.marshal(marshallDataFormat);
+ rdInTemplateSource = rdInTemplateSource.marshal("{{marshall}}");
}
if (!ObjectHelper.isEmpty(unmarshallDataFormat)) {
- rdInTemplateSource = rdInTemplateSource.unmarshal(unmarshallDataFormat);
+ rdInTemplateSource = rdInTemplateSource.unmarshal("{{unmarshall}}");
}
if (getContext().getRegistry().lookupByName("aggregate") != null) {
diff --git a/core/src/test/java/org/apache/camel/kafkaconnector/CamelSourceTaskTest.java b/core/src/test/java/org/apache/camel/kafkaconnector/CamelSourceTaskTest.java
index b1271ac..36ae9e2 100644
--- a/core/src/test/java/org/apache/camel/kafkaconnector/CamelSourceTaskTest.java
+++ b/core/src/test/java/org/apache/camel/kafkaconnector/CamelSourceTaskTest.java
@@ -235,9 +235,6 @@ public class CamelSourceTaskTest {
CamelSourceTask sourceTask = new CamelSourceTask();
sourceTask.start(props);
-// assertEquals(4, sourceTask.getCms().getCamelContext().getEndpoints().size());
-
-
sourceTask.getCms().getCamelContext().getEndpoints().stream()
.filter(e -> e.getEndpointUri().startsWith("timer"))
.forEach(e -> {
@@ -261,8 +258,6 @@ public class CamelSourceTaskTest {
CamelSourceTask sourceTask = new CamelSourceTask();
sourceTask.start(props);
-// assertEquals(4, sourceTask.getCms().getCamelContext().getEndpoints().size());
-
sourceTask.getCms().getCamelContext().getEndpoints().stream()
.filter(e -> e.getEndpointUri().startsWith("seda"))
.forEach(e -> {