Kafka Source #247

docs/modules/ROOT/pages/kafka-source.adoc

@@ -15,6 +15,10 @@ The following table summarizes the configuration options available for the `kafk
 | *password {empty}* *| Password| Password to authenticate to kafka| string| | 
 | *topic {empty}* *| Topic Names| Comma separated list of Kafka topic names| string| | 
 | *username {empty}* *| Username| Username to authenticate to Kafka| string| | 
+| allowManualCommit| Allow Manual Commit| Whether to allow doing manual commits| boolean| `false`| 
+| autoCommitEnable| Auto Commit Enable| If true, periodically commit to ZooKeeper the offset of messages already fetched by the consumer| boolean| `true`| 
+| autoOffsetReset| Auto Offset Reset| What to do when there is no initial offset. There are 3 enums and the value can be one of latest, earliest, none| string| `"latest"`| 
+| pollOnError| Poll On Error Behavior| 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| string| `"ERROR_HANDLER"`| 
 | saslMechanism| SASL Mechanism| The Simple Authentication and Security Layer (SASL) Mechanism used.| string| `"PLAIN"`| 
 | securityProtocol| Security Protocol| Protocol used to communicate with brokers. SASL_PLAINTEXT, PLAINTEXT, SASL_SSL and SSL are supported| string| `"SASL_SSL"`| 
 |===

kafka-source.kamelet.yaml

@@ -49,6 +49,30 @@ spec:
         format: password
         x-descriptors:
         - urn:alm:descriptor:com.tectonic.ui:password
+      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
+        x-descriptors:
+        - 'urn:alm:descriptor:com.tectonic.ui:checkbox'
+      allowManualCommit:
+        title: Allow Manual Commit
+        description: Whether to allow doing manual commits
+        type: boolean
+        default: false
+        x-descriptors:
+        - 'urn:alm:descriptor:com.tectonic.ui:checkbox'
+      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"
   dependencies:
     - "camel:kafka"
     - "camel:kamelet"
@@ -60,5 +84,9 @@ spec:
         securityProtocol: "{{securityProtocol}}"
         saslMechanism: "{{saslMechanism}}"
         saslJaasConfig: "org.apache.kafka.common.security.plain.PlainLoginModule required username='{{username}}' password='{{password}}';"
+        autoCommitEnable: "{{autoCommitEnable}}"
+        allowManualCommit: "{{allowManualCommit}}"
+        pollOnError: "{{pollOnError}}"
+        autoOffsetReset: "{{autoOffsetReset}}"
       steps:
       - to: "kamelet:sink"

library/camel-kamelets-catalog/src/main/resources/kamelets/kafka-source.kamelet.yaml

@@ -49,6 +49,30 @@ spec:
         format: password
         x-descriptors:
         - urn:alm:descriptor:com.tectonic.ui:password
+      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
+        x-descriptors:
+        - 'urn:alm:descriptor:com.tectonic.ui:checkbox'
+      allowManualCommit:
+        title: Allow Manual Commit
+        description: Whether to allow doing manual commits
+        type: boolean
+        default: false
+        x-descriptors:
+        - 'urn:alm:descriptor:com.tectonic.ui:checkbox'
+      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"
   dependencies:
     - "camel:kafka"
     - "camel:kamelet"
@@ -60,5 +84,9 @@ spec:
         securityProtocol: "{{securityProtocol}}"
         saslMechanism: "{{saslMechanism}}"
         saslJaasConfig: "org.apache.kafka.common.security.plain.PlainLoginModule required username='{{username}}' password='{{password}}';"
+        autoCommitEnable: "{{autoCommitEnable}}"
+        allowManualCommit: "{{allowManualCommit}}"
+        pollOnError: "{{pollOnError}}"
+        autoOffsetReset: "{{autoOffsetReset}}"
       steps:
       - to: "kamelet:sink"