You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@camel.apache.org by ac...@apache.org on 2024/01/30 10:43:35 UTC
(camel-kamelets) branch main updated (49a16878 -> 148c476f)
This is an automated email from the ASF dual-hosted git repository.
acosentino pushed a change to branch main
in repository https://gitbox.apache.org/repos/asf/camel-kamelets.git
from 49a16878 Bump org.apache.rat:apache-rat-plugin from 0.16 to 0.16.1
new d67c06d2 Support Kafka Batch as Kamelet source - Kafka Not Secured Source as batch
new 6a1101f5 Support Kafka Batch as Kamelet source - Kafka Not Secured Source as batch
new 6799eed1 Support Kafka Batch as Kamelet source - Kafka Batch Commit action
new 3c2053ba Support Kafka Batch as Kamelet source - Kafka Batch Commit action
new e8f910f2 Support Kafka Batch as Kamelet source - Kafka Not Secured Source as batch
new 9be1caa4 Support Kafka Batch as Kamelet source - Kafka Not Secured Source as batch
new 148c476f Support Kafka Batch as Kamelet source - Kafka Not Secured Source as batch
The 7 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:
docs/modules/ROOT/nav.adoc | 2 ++
... kafka-batch-manual-commit-action.kamelet.yaml} | 8 ++++----
.../kafka-batch-not-secured-source.kamelet.yaml | 22 +++++++++++++++++++---
.../kamelets/catalog/KameletsCatalogTest.java | 12 ++++++------
.../{ManualCommit.java => BatchManualCommit.java} | 17 +++++++++++++----
.../kafka-batch-manual-commit-action.kamelet.yaml | 8 ++++----
...=> kafka-batch-not-secured-source.kamelet.yaml} | 22 +++++++++++++++++++---
7 files changed, 67 insertions(+), 24 deletions(-)
copy kamelets/{kafka-manual-commit-action.kamelet.yaml => kafka-batch-manual-commit-action.kamelet.yaml} (95%)
copy library/camel-kamelets/src/main/resources/kamelets/kafka-not-secured-source.kamelet.yaml => kamelets/kafka-batch-not-secured-source.kamelet.yaml (88%)
copy library/camel-kamelets-utils/src/main/java/org/apache/camel/kamelets/utils/transform/kafka/{ManualCommit.java => BatchManualCommit.java} (67%)
copy kamelets/kafka-manual-commit-action.kamelet.yaml => library/camel-kamelets/src/main/resources/kamelets/kafka-batch-manual-commit-action.kamelet.yaml (95%)
copy library/camel-kamelets/src/main/resources/kamelets/{kafka-not-secured-source.kamelet.yaml => kafka-batch-not-secured-source.kamelet.yaml} (88%)
(camel-kamelets) 01/07: Support Kafka Batch as Kamelet source - Kafka Not Secured Source as batch
Posted by ac...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
acosentino pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/camel-kamelets.git
commit d67c06d291024de03c48ca2766a639430da9026a
Author: Andrea Cosentino <an...@gmail.com>
AuthorDate: Tue Jan 30 11:28:24 2024 +0100
Support Kafka Batch as Kamelet source - Kafka Not Secured Source as batch
Signed-off-by: Andrea Cosentino <an...@gmail.com>
---
.../kafka-batch-not-secured-source.kamelet.yaml | 129 +++++++++++++++++++++
.../utils/transform/kafka/BatchManualCommit.java | 42 +++++++
2 files changed, 171 insertions(+)
diff --git a/kamelets/kafka-batch-not-secured-source.kamelet.yaml b/kamelets/kafka-batch-not-secured-source.kamelet.yaml
new file mode 100644
index 00000000..14cb1040
--- /dev/null
+++ b/kamelets/kafka-batch-not-secured-source.kamelet.yaml
@@ -0,0 +1,129 @@
+# ---------------------------------------------------------------------------
+# 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.
+# ---------------------------------------------------------------------------
+apiVersion: camel.apache.org/v1
+kind: Kamelet
+metadata:
+ name: kafka-batch-not-secured-source
+ annotations:
+ camel.apache.org/kamelet.support.level: "Stable"
+ camel.apache.org/catalog.version: "4.4.0-SNAPSHOT"
+ camel.apache.org/kamelet.icon: "data:image/svg+xml;base64,PD94bWwgdmVyc2lvbj0iMS4wIiBlbmNvZGluZz0idXRmLTgiPz4NCjwhLS0gR2VuZXJhdG9yOiBBZG9iZSBJbGx1c3RyYXRvciAxOS4wLjAsIFNWRyBFeHBvcnQgUGx1Zy1JbiAuIFNWRyBWZXJzaW9uOiA2LjAwIEJ1aWxkIDApICAtLT4NCjxzdmcgdmVyc2lvbj0iMS4xIiBpZD0iTGF5ZXJfMSIgeG1sbnM9Imh0dHA6Ly93d3cudzMub3JnLzIwMDAvc3ZnIiB4bWxuczp4bGluaz0iaHR0cDovL3d3dy53My5vcmcvMTk5OS94bGluayIgeD0iMHB4IiB5PSIwcHgiDQoJIHZpZXdCb3g9IjAgMCA1MDAgNTAwIiBzdHlsZT0iZW5hYmxlLWJhY2tncm91bmQ6bmV3IDAgMCA1MD [...]
+ camel.apache.org/provider: "Apache Software Foundation"
+ camel.apache.org/kamelet.group: "Kafka"
+ camel.apache.org/kamelet.namespace: "Kafka"
+ camel.apache.org/keda.type: "kafka"
+ labels:
+ camel.apache.org/kamelet.type: "source"
+spec:
+ definition:
+ title: "Kafka Batch Not Secured Source"
+ description: |-
+ Receive data from Kafka topics in batch on an insecure broker and commit them manually through KafkaManualCommit.
+ required:
+ - topic
+ - bootstrapServers
+ type: object
+ properties:
+ topic:
+ title: Topic Names
+ description: Comma separated list of Kafka topic names
+ type: string
+ x-descriptors:
+ - urn:keda:metadata:topic
+ - urn:keda:required
+ bootstrapServers:
+ title: Bootstrap Servers
+ description: Comma separated list of Kafka Broker URLs
+ type: string
+ x-descriptors:
+ - urn:keda:metadata:bootstrapServers
+ - urn:keda:required
+ autoCommitEnable:
+ title: Auto Commit Enable
+ description: If true, periodically commit to ZooKeeper the offset of messages already fetched by the consumer
+ type: boolean
+ default: true
+ allowManualCommit:
+ title: Allow Manual Commit
+ description: Whether to allow doing manual commits
+ type: boolean
+ default: false
+ pollOnError:
+ title: Poll On Error Behavior
+ description: What to do if kafka threw an exception while polling for new messages. There are 5 enums and the value can be one of DISCARD, ERROR_HANDLER, RECONNECT, RETRY, STOP
+ type: string
+ default: "ERROR_HANDLER"
+ autoOffsetReset:
+ title: Auto Offset Reset
+ description: What to do when there is no initial offset. There are 3 enums and the value can be one of latest, earliest, none
+ type: string
+ default: "latest"
+ x-descriptors:
+ - urn:keda:metadata:offsetResetPolicy
+ consumerGroup:
+ title: Consumer Group
+ description: A string that uniquely identifies the group of consumers to which this source belongs
+ type: string
+ example: "my-group-id"
+ x-descriptors:
+ - urn:keda:metadata:consumerGroup
+ - urn:keda:required
+ deserializeHeaders:
+ title: Automatically Deserialize Headers
+ description: When enabled the Kamelet source will deserialize all message headers to String representation.
+ type: boolean
+ default: true
+ batchSize:
+ title: The batch dimension
+ description: The maximum number of records returned in a single call to poll()
+ type: int
+ default: 500
+ pollTimeout:
+ title: The poll timeout interval
+ description: The timeout used when polling the KafkaConsumer
+ type: int
+ default: 5000
+ dependencies:
+ - "mvn:org.apache.camel.kamelets:camel-kamelets-utils:4.4.0-SNAPSHOT"
+ - "camel:kafka"
+ - "camel:core"
+ - "camel:kamelet"
+ template:
+ beans:
+ - name: kafkaHeaderDeserializer
+ type: "#class:org.apache.camel.kamelets.utils.serialization.kafka.KafkaHeaderDeserializer"
+ properties:
+ enabled: '{{deserializeHeaders}}'
+ - name: manualCommitFactory
+ type: "#class:org.apache.camel.component.kafka.consumer.DefaultKafkaManualCommitFactory"
+ from:
+ uri: "kafka:{{topic}}"
+ parameters:
+ brokers: "{{bootstrapServers}}"
+ autoCommitEnable: "{{autoCommitEnable}}"
+ allowManualCommit: "{{allowManualCommit}}"
+ pollOnError: "{{pollOnError}}"
+ autoOffsetReset: "{{autoOffsetReset}}"
+ groupId: "{{?consumerGroup}}"
+ maxPollRecords: "{{batchSize}}"
+ pollTimeoutMs: "{{pollTimeout}}"
+ batching: true
+ kafkaManualCommitFactory: "#bean:{{manualCommitFactory}}"
+ steps:
+ - process:
+ ref: "{{kafkaHeaderDeserializer}}"
+ - to: "kamelet:sink"
diff --git a/library/camel-kamelets-utils/src/main/java/org/apache/camel/kamelets/utils/transform/kafka/BatchManualCommit.java b/library/camel-kamelets-utils/src/main/java/org/apache/camel/kamelets/utils/transform/kafka/BatchManualCommit.java
new file mode 100644
index 00000000..f6d7d841
--- /dev/null
+++ b/library/camel-kamelets-utils/src/main/java/org/apache/camel/kamelets/utils/transform/kafka/BatchManualCommit.java
@@ -0,0 +1,42 @@
+/*
+ * 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.kamelets.utils.transform.kafka;
+
+import org.apache.camel.Exchange;
+import org.apache.camel.Processor;
+import org.apache.camel.component.kafka.KafkaConstants;
+import org.apache.camel.component.kafka.consumer.KafkaManualCommit;
+
+import java.util.List;
+
+public class BatchManualCommit implements Processor {
+
+ @Override
+ public void process(Exchange exchange) throws Exception {
+ List<?> exchanges = exchange.getMessage().getBody(List.class);
+ if (exchanges.size() > 0) {
+ final Object tmp = exchanges.get(exchanges.size() - 1);
+ if (tmp instanceof Exchange element) {
+ KafkaManualCommit manual
+ = element.getMessage().getHeader(KafkaConstants.MANUAL_COMMIT, KafkaManualCommit.class);
+ if (manual != null) {
+ manual.commit();
+ }
+ }
+ }
+ }
+}
(camel-kamelets) 05/07: Support Kafka Batch as Kamelet source - Kafka Not Secured Source as batch
Posted by ac...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
acosentino pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/camel-kamelets.git
commit e8f910f223f602b6fef4b5ff93842cf67fbdd294
Author: Andrea Cosentino <an...@gmail.com>
AuthorDate: Tue Jan 30 11:32:00 2024 +0100
Support Kafka Batch as Kamelet source - Kafka Not Secured Source as batch
Signed-off-by: Andrea Cosentino <an...@gmail.com>
---
.../resources/kamelets/kafka-batch-not-secured-source.kamelet.yaml | 5 ++---
1 file changed, 2 insertions(+), 3 deletions(-)
diff --git a/library/camel-kamelets/src/main/resources/kamelets/kafka-batch-not-secured-source.kamelet.yaml b/library/camel-kamelets/src/main/resources/kamelets/kafka-batch-not-secured-source.kamelet.yaml
index bef14b00..14cb1040 100644
--- a/library/camel-kamelets/src/main/resources/kamelets/kafka-batch-not-secured-source.kamelet.yaml
+++ b/library/camel-kamelets/src/main/resources/kamelets/kafka-batch-not-secured-source.kamelet.yaml
@@ -32,7 +32,7 @@ spec:
definition:
title: "Kafka Batch Not Secured Source"
description: |-
- Receive data from Kafka topics in batch on an insecure broker.
+ Receive data from Kafka topics in batch on an insecure broker and commit them manually through KafkaManualCommit.
required:
- topic
- bootstrapServers
@@ -95,7 +95,7 @@ spec:
pollTimeout:
title: The poll timeout interval
description: The timeout used when polling the KafkaConsumer
- type: long
+ type: int
default: 5000
dependencies:
- "mvn:org.apache.camel.kamelets:camel-kamelets-utils:4.4.0-SNAPSHOT"
@@ -122,7 +122,6 @@ spec:
maxPollRecords: "{{batchSize}}"
pollTimeoutMs: "{{pollTimeout}}"
batching: true
- synchronous: true
kafkaManualCommitFactory: "#bean:{{manualCommitFactory}}"
steps:
- process:
(camel-kamelets) 06/07: Support Kafka Batch as Kamelet source - Kafka Not Secured Source as batch
Posted by ac...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
acosentino pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/camel-kamelets.git
commit 9be1caa41060b05991833df0b2fb8720e6f73e08
Author: Andrea Cosentino <an...@gmail.com>
AuthorDate: Tue Jan 30 11:35:35 2024 +0100
Support Kafka Batch as Kamelet source - Kafka Not Secured Source as batch
Signed-off-by: Andrea Cosentino <an...@gmail.com>
---
.../apache/camel/kamelets/catalog/KameletsCatalogTest.java | 12 ++++++------
1 file changed, 6 insertions(+), 6 deletions(-)
diff --git a/library/camel-kamelets-catalog/src/test/java/org/apache/camel/kamelets/catalog/KameletsCatalogTest.java b/library/camel-kamelets-catalog/src/test/java/org/apache/camel/kamelets/catalog/KameletsCatalogTest.java
index 993a7fd6..3e82a7fb 100644
--- a/library/camel-kamelets-catalog/src/test/java/org/apache/camel/kamelets/catalog/KameletsCatalogTest.java
+++ b/library/camel-kamelets-catalog/src/test/java/org/apache/camel/kamelets/catalog/KameletsCatalogTest.java
@@ -189,12 +189,12 @@ public class KameletsCatalogTest {
verifyHeaders("jira-add-issue-sink", 17);
verifyHeaders("jira-source", 3);
verifyHeaders("jira-oauth-source", 3);
- verifyHeaders("jms-amqp-10-source", 14);
- verifyHeaders("jms-amqp-10-sink", 17);
- verifyHeaders("jms-apache-artemis-source", 14);
- verifyHeaders("jms-apache-artemis-sink", 17);
- verifyHeaders("jms-ibm-mq-source", 14);
- verifyHeaders("jms-ibm-mq-sink", 17);
+ verifyHeaders("jms-amqp-10-source", 15);
+ verifyHeaders("jms-amqp-10-sink", 18);
+ verifyHeaders("jms-apache-artemis-source", 15);
+ verifyHeaders("jms-apache-artemis-sink", 18);
+ verifyHeaders("jms-ibm-mq-source", 15);
+ verifyHeaders("jms-ibm-mq-sink", 18);
verifyHeaders("kafka-source", 9);
verifyHeaders("kafka-sink", 5);
verifyHeaders("kafka-ssl-source", 9);
(camel-kamelets) 03/07: Support Kafka Batch as Kamelet source - Kafka Batch Commit action
Posted by ac...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
acosentino pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/camel-kamelets.git
commit 6799eed1d3b3550b32d2db5f04d0c6dd749b8633
Author: Andrea Cosentino <an...@gmail.com>
AuthorDate: Tue Jan 30 11:29:32 2024 +0100
Support Kafka Batch as Kamelet source - Kafka Batch Commit action
Signed-off-by: Andrea Cosentino <an...@gmail.com>
---
.../kafka-batch-manual-commit-action.kamelet.yaml | 42 ++++++++++++++++++++++
1 file changed, 42 insertions(+)
diff --git a/kamelets/kafka-batch-manual-commit-action.kamelet.yaml b/kamelets/kafka-batch-manual-commit-action.kamelet.yaml
new file mode 100644
index 00000000..12827e06
--- /dev/null
+++ b/kamelets/kafka-batch-manual-commit-action.kamelet.yaml
@@ -0,0 +1,42 @@
+# ---------------------------------------------------------------------------
+# 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.
+# ---------------------------------------------------------------------------
+apiVersion: camel.apache.org/v1
+kind: Kamelet
+metadata:
+ name: kafka-batch-manual-commit-action
+ labels:
+ camel.apache.org/kamelet.type: "action"
+ annotations:
+ camel.apache.org/kamelet.support.level: "Stable"
+ camel.apache.org/catalog.version: "4.4.0-SNAPSHOT"
+ camel.apache.org/kamelet.icon: "data:image/svg+xml;base64,PD94bWwgdmVyc2lvbj0iMS4wIiBlbmNvZGluZz0iVVRGLTgiIHN0YW5kYWxvbmU9Im5vIj8+CjxzdmcKICAgeG1sbnM6ZGM9Imh0dHA6Ly9wdXJsLm9yZy9kYy9lbGVtZW50cy8xLjEvIgogICB4bWxuczpjYz0iaHR0cDovL2NyZWF0aXZlY29tbW9ucy5vcmcvbnMjIgogICB4bWxuczpyZGY9Imh0dHA6Ly93d3cudzMub3JnLzE5OTkvMDIvMjItcmRmLXN5bnRheC1ucyMiCiAgIHhtbG5zOnN2Zz0iaHR0cDovL3d3dy53My5vcmcvMjAwMC9zdmciCiAgIHhtbG5zPSJodHRwOi8vd3d3LnczLm9yZy8yMDAwL3N2ZyIKICAgeG1sbnM6c29kaXBvZGk9Imh0dHA6Ly9zb2RpcG [...]
+ camel.apache.org/provider: "Apache Software Foundation"
+ camel.apache.org/kamelet.group: "Actions"
+ camel.apache.org/kamelet.namespace: "Kafka"
+spec:
+ definition:
+ title: "Kafka Batch Manual Commit Action"
+ description: "Manually commit Kafka Batch Offset"
+ type: object
+ dependencies:
+ - "mvn:org.apache.camel.kamelets:camel-kamelets-utils:4.4.0-SNAPSHOT"
+ - "camel:kamelet"
+ template:
+ from:
+ uri: kamelet:source
+ steps:
+ - bean: "org.apache.camel.kamelets.utils.transform.kafka.BatchManualCommit"
(camel-kamelets) 07/07: Support Kafka Batch as Kamelet source - Kafka Not Secured Source as batch
Posted by ac...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
acosentino pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/camel-kamelets.git
commit 148c476f3b2e952b4c0bcb40765a3c0a00ce2a20
Author: Andrea Cosentino <an...@gmail.com>
AuthorDate: Tue Jan 30 11:39:50 2024 +0100
Support Kafka Batch as Kamelet source - Kafka Not Secured Source as batch
Signed-off-by: Andrea Cosentino <an...@gmail.com>
---
docs/modules/ROOT/nav.adoc | 2 ++
kamelets/kafka-batch-not-secured-source.kamelet.yaml | 4 ++--
2 files changed, 4 insertions(+), 2 deletions(-)
diff --git a/docs/modules/ROOT/nav.adoc b/docs/modules/ROOT/nav.adoc
index e1d902e6..d7be04ed 100644
--- a/docs/modules/ROOT/nav.adoc
+++ b/docs/modules/ROOT/nav.adoc
@@ -131,6 +131,8 @@
* xref:kafka-apicurio-registry-not-secured-source.adoc[]
* xref:kafka-azure-schema-registry-sink.adoc[]
* xref:kafka-azure-schema-registry-source.adoc[]
+* xref:kafka-batch-manual-commit-action.adoc[]
+* xref:kafka-batch-not-secured-source.adoc[]
* xref:kafka-manual-commit-action.adoc[]
* xref:kafka-not-secured-sink.adoc[]
* xref:kafka-not-secured-source.adoc[]
diff --git a/kamelets/kafka-batch-not-secured-source.kamelet.yaml b/kamelets/kafka-batch-not-secured-source.kamelet.yaml
index 14cb1040..15b6525b 100644
--- a/kamelets/kafka-batch-not-secured-source.kamelet.yaml
+++ b/kamelets/kafka-batch-not-secured-source.kamelet.yaml
@@ -88,12 +88,12 @@ spec:
type: boolean
default: true
batchSize:
- title: The batch dimension
+ title: Batch Dimension
description: The maximum number of records returned in a single call to poll()
type: int
default: 500
pollTimeout:
- title: The poll timeout interval
+ title: Poll Timeout Interval
description: The timeout used when polling the KafkaConsumer
type: int
default: 5000
(camel-kamelets) 04/07: Support Kafka Batch as Kamelet source - Kafka Batch Commit action
Posted by ac...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
acosentino pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/camel-kamelets.git
commit 3c2053ba46e2643fe5aeb4aebeed83b8fc917e86
Author: Andrea Cosentino <an...@gmail.com>
AuthorDate: Tue Jan 30 11:31:16 2024 +0100
Support Kafka Batch as Kamelet source - Kafka Batch Commit action
Signed-off-by: Andrea Cosentino <an...@gmail.com>
---
.../kafka-batch-manual-commit-action.kamelet.yaml | 42 ++++++++++++++++++++++
1 file changed, 42 insertions(+)
diff --git a/library/camel-kamelets/src/main/resources/kamelets/kafka-batch-manual-commit-action.kamelet.yaml b/library/camel-kamelets/src/main/resources/kamelets/kafka-batch-manual-commit-action.kamelet.yaml
new file mode 100644
index 00000000..12827e06
--- /dev/null
+++ b/library/camel-kamelets/src/main/resources/kamelets/kafka-batch-manual-commit-action.kamelet.yaml
@@ -0,0 +1,42 @@
+# ---------------------------------------------------------------------------
+# 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.
+# ---------------------------------------------------------------------------
+apiVersion: camel.apache.org/v1
+kind: Kamelet
+metadata:
+ name: kafka-batch-manual-commit-action
+ labels:
+ camel.apache.org/kamelet.type: "action"
+ annotations:
+ camel.apache.org/kamelet.support.level: "Stable"
+ camel.apache.org/catalog.version: "4.4.0-SNAPSHOT"
+ camel.apache.org/kamelet.icon: "data:image/svg+xml;base64,PD94bWwgdmVyc2lvbj0iMS4wIiBlbmNvZGluZz0iVVRGLTgiIHN0YW5kYWxvbmU9Im5vIj8+CjxzdmcKICAgeG1sbnM6ZGM9Imh0dHA6Ly9wdXJsLm9yZy9kYy9lbGVtZW50cy8xLjEvIgogICB4bWxuczpjYz0iaHR0cDovL2NyZWF0aXZlY29tbW9ucy5vcmcvbnMjIgogICB4bWxuczpyZGY9Imh0dHA6Ly93d3cudzMub3JnLzE5OTkvMDIvMjItcmRmLXN5bnRheC1ucyMiCiAgIHhtbG5zOnN2Zz0iaHR0cDovL3d3dy53My5vcmcvMjAwMC9zdmciCiAgIHhtbG5zPSJodHRwOi8vd3d3LnczLm9yZy8yMDAwL3N2ZyIKICAgeG1sbnM6c29kaXBvZGk9Imh0dHA6Ly9zb2RpcG [...]
+ camel.apache.org/provider: "Apache Software Foundation"
+ camel.apache.org/kamelet.group: "Actions"
+ camel.apache.org/kamelet.namespace: "Kafka"
+spec:
+ definition:
+ title: "Kafka Batch Manual Commit Action"
+ description: "Manually commit Kafka Batch Offset"
+ type: object
+ dependencies:
+ - "mvn:org.apache.camel.kamelets:camel-kamelets-utils:4.4.0-SNAPSHOT"
+ - "camel:kamelet"
+ template:
+ from:
+ uri: kamelet:source
+ steps:
+ - bean: "org.apache.camel.kamelets.utils.transform.kafka.BatchManualCommit"
(camel-kamelets) 02/07: Support Kafka Batch as Kamelet source - Kafka Not Secured Source as batch
Posted by ac...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
acosentino pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/camel-kamelets.git
commit 6a1101f545e3072d264aec7ed093eccc6f740624
Author: Andrea Cosentino <an...@gmail.com>
AuthorDate: Tue Jan 30 11:29:13 2024 +0100
Support Kafka Batch as Kamelet source - Kafka Not Secured Source as batch
Signed-off-by: Andrea Cosentino <an...@gmail.com>
---
.../kafka-batch-not-secured-source.kamelet.yaml | 130 +++++++++++++++++++++
1 file changed, 130 insertions(+)
diff --git a/library/camel-kamelets/src/main/resources/kamelets/kafka-batch-not-secured-source.kamelet.yaml b/library/camel-kamelets/src/main/resources/kamelets/kafka-batch-not-secured-source.kamelet.yaml
new file mode 100644
index 00000000..bef14b00
--- /dev/null
+++ b/library/camel-kamelets/src/main/resources/kamelets/kafka-batch-not-secured-source.kamelet.yaml
@@ -0,0 +1,130 @@
+# ---------------------------------------------------------------------------
+# 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.
+# ---------------------------------------------------------------------------
+apiVersion: camel.apache.org/v1
+kind: Kamelet
+metadata:
+ name: kafka-batch-not-secured-source
+ annotations:
+ camel.apache.org/kamelet.support.level: "Stable"
+ camel.apache.org/catalog.version: "4.4.0-SNAPSHOT"
+ camel.apache.org/kamelet.icon: "data:image/svg+xml;base64,PD94bWwgdmVyc2lvbj0iMS4wIiBlbmNvZGluZz0idXRmLTgiPz4NCjwhLS0gR2VuZXJhdG9yOiBBZG9iZSBJbGx1c3RyYXRvciAxOS4wLjAsIFNWRyBFeHBvcnQgUGx1Zy1JbiAuIFNWRyBWZXJzaW9uOiA2LjAwIEJ1aWxkIDApICAtLT4NCjxzdmcgdmVyc2lvbj0iMS4xIiBpZD0iTGF5ZXJfMSIgeG1sbnM9Imh0dHA6Ly93d3cudzMub3JnLzIwMDAvc3ZnIiB4bWxuczp4bGluaz0iaHR0cDovL3d3dy53My5vcmcvMTk5OS94bGluayIgeD0iMHB4IiB5PSIwcHgiDQoJIHZpZXdCb3g9IjAgMCA1MDAgNTAwIiBzdHlsZT0iZW5hYmxlLWJhY2tncm91bmQ6bmV3IDAgMCA1MD [...]
+ camel.apache.org/provider: "Apache Software Foundation"
+ camel.apache.org/kamelet.group: "Kafka"
+ camel.apache.org/kamelet.namespace: "Kafka"
+ camel.apache.org/keda.type: "kafka"
+ labels:
+ camel.apache.org/kamelet.type: "source"
+spec:
+ definition:
+ title: "Kafka Batch Not Secured Source"
+ description: |-
+ Receive data from Kafka topics in batch on an insecure broker.
+ required:
+ - topic
+ - bootstrapServers
+ type: object
+ properties:
+ topic:
+ title: Topic Names
+ description: Comma separated list of Kafka topic names
+ type: string
+ x-descriptors:
+ - urn:keda:metadata:topic
+ - urn:keda:required
+ bootstrapServers:
+ title: Bootstrap Servers
+ description: Comma separated list of Kafka Broker URLs
+ type: string
+ x-descriptors:
+ - urn:keda:metadata:bootstrapServers
+ - urn:keda:required
+ autoCommitEnable:
+ title: Auto Commit Enable
+ description: If true, periodically commit to ZooKeeper the offset of messages already fetched by the consumer
+ type: boolean
+ default: true
+ allowManualCommit:
+ title: Allow Manual Commit
+ description: Whether to allow doing manual commits
+ type: boolean
+ default: false
+ pollOnError:
+ title: Poll On Error Behavior
+ description: What to do if kafka threw an exception while polling for new messages. There are 5 enums and the value can be one of DISCARD, ERROR_HANDLER, RECONNECT, RETRY, STOP
+ type: string
+ default: "ERROR_HANDLER"
+ autoOffsetReset:
+ title: Auto Offset Reset
+ description: What to do when there is no initial offset. There are 3 enums and the value can be one of latest, earliest, none
+ type: string
+ default: "latest"
+ x-descriptors:
+ - urn:keda:metadata:offsetResetPolicy
+ consumerGroup:
+ title: Consumer Group
+ description: A string that uniquely identifies the group of consumers to which this source belongs
+ type: string
+ example: "my-group-id"
+ x-descriptors:
+ - urn:keda:metadata:consumerGroup
+ - urn:keda:required
+ deserializeHeaders:
+ title: Automatically Deserialize Headers
+ description: When enabled the Kamelet source will deserialize all message headers to String representation.
+ type: boolean
+ default: true
+ batchSize:
+ title: The batch dimension
+ description: The maximum number of records returned in a single call to poll()
+ type: int
+ default: 500
+ pollTimeout:
+ title: The poll timeout interval
+ description: The timeout used when polling the KafkaConsumer
+ type: long
+ default: 5000
+ dependencies:
+ - "mvn:org.apache.camel.kamelets:camel-kamelets-utils:4.4.0-SNAPSHOT"
+ - "camel:kafka"
+ - "camel:core"
+ - "camel:kamelet"
+ template:
+ beans:
+ - name: kafkaHeaderDeserializer
+ type: "#class:org.apache.camel.kamelets.utils.serialization.kafka.KafkaHeaderDeserializer"
+ properties:
+ enabled: '{{deserializeHeaders}}'
+ - name: manualCommitFactory
+ type: "#class:org.apache.camel.component.kafka.consumer.DefaultKafkaManualCommitFactory"
+ from:
+ uri: "kafka:{{topic}}"
+ parameters:
+ brokers: "{{bootstrapServers}}"
+ autoCommitEnable: "{{autoCommitEnable}}"
+ allowManualCommit: "{{allowManualCommit}}"
+ pollOnError: "{{pollOnError}}"
+ autoOffsetReset: "{{autoOffsetReset}}"
+ groupId: "{{?consumerGroup}}"
+ maxPollRecords: "{{batchSize}}"
+ pollTimeoutMs: "{{pollTimeout}}"
+ batching: true
+ synchronous: true
+ kafkaManualCommitFactory: "#bean:{{manualCommitFactory}}"
+ steps:
+ - process:
+ ref: "{{kafkaHeaderDeserializer}}"
+ - to: "kamelet:sink"