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:35:51 UTC
(camel-kamelets) branch kafka-batch-source created (now 86e7cdca)
This is an automated email from the ASF dual-hosted git repository.
acosentino pushed a change to branch kafka-batch-source
in repository https://gitbox.apache.org/repos/asf/camel-kamelets.git
at 86e7cdca Support Kafka Batch as Kamelet source - Kafka Not Secured Source as batch
This branch includes the following new commits:
new cb254541 Support Kafka Batch as Kamelet source - Kafka Not Secured Source as batch
new 3f9fea7a Support Kafka Batch as Kamelet source - Kafka Not Secured Source as batch
new da19628a Support Kafka Batch as Kamelet source - Kafka Batch Commit action
new 69f24363 Support Kafka Batch as Kamelet source - Kafka Batch Commit action
new 6cf7636f Support Kafka Batch as Kamelet source - Kafka Not Secured Source as batch
new 86e7cdca Support Kafka Batch as Kamelet source - Kafka Not Secured Source as batch
The 6 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.
(camel-kamelets) 05/06: 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 kafka-batch-source
in repository https://gitbox.apache.org/repos/asf/camel-kamelets.git
commit 6cf7636f60ba117be8ad1241e946eecabbc3aa05
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) 02/06: 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 kafka-batch-source
in repository https://gitbox.apache.org/repos/asf/camel-kamelets.git
commit 3f9fea7ac0e3cdbe6135167624500f68e6f9c67b
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"
(camel-kamelets) 04/06: 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 kafka-batch-source
in repository https://gitbox.apache.org/repos/asf/camel-kamelets.git
commit 69f24363100042ae487343556e1f84a35aac8469
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) 01/06: 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 kafka-batch-source
in repository https://gitbox.apache.org/repos/asf/camel-kamelets.git
commit cb254541492e8582d639df4732cf64e159ad6a82
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) 03/06: 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 kafka-batch-source
in repository https://gitbox.apache.org/repos/asf/camel-kamelets.git
commit da19628acda9f23af525c9d38a6da7ecb13a6879
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) 06/06: 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 kafka-batch-source
in repository https://gitbox.apache.org/repos/asf/camel-kamelets.git
commit 86e7cdca03288f28ff589465887f98966c6c6175
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);