Repository: apache/camel-k
Pull request #2266 (2021-05-05 09:06:12)

[Backport 1.4.x] Kamelet Binding error handler
<!-- Description -->




<!--
Enter your extended release note in the below block. If the PR requires
additional action from users switching to the new release, include the string
"action required". If no release note is required, write "NONE". 

You can (optionally) mark this PR with labels "kind/bug" or "kind/feature" to make sure
the text is added to the right section of the release notes. 
-->

**Release Note**
```release-note
feat: Kamelet Binding error handler
```



Patch:
diff --git a/config/crd/bases/camel.apache.org_kameletbindings.yaml b/config/crd/bases/camel.apache.org_kameletbindings.yaml
index c9a8b8724..60ac3ba49 100644
--- a/config/crd/bases/camel.apache.org_kameletbindings.yaml
+++ b/config/crd/bases/camel.apache.org_kameletbindings.yaml
@@ -63,6 +63,11 @@ spec:
           spec:
             description: KameletBindingSpec --
             properties:
+              errorHandler:
+                description: ErrorHandler is an optional handler called upon an error
+                  occuring in the integration
+                type: object
+                x-kubernetes-preserve-unknown-fields: true
               integration:
                 description: Integration is an optional integration used to specify
                   custom parameters
diff --git a/deploy/olm-catalog/camel-k-dev/1.4.0/camel.apache.org_kameletbindings.yaml b/deploy/olm-catalog/camel-k-dev/1.4.0/camel.apache.org_kameletbindings.yaml
index c9a8b8724..2b78d2f58 100644
--- a/deploy/olm-catalog/camel-k-dev/1.4.0/camel.apache.org_kameletbindings.yaml
+++ b/deploy/olm-catalog/camel-k-dev/1.4.0/camel.apache.org_kameletbindings.yaml
@@ -63,6 +63,11 @@ spec:
           spec:
             description: KameletBindingSpec --
             properties:
+              errorHandler:
+                description: ErrorHandler is an optional handler called upon an error
+                  occuring in the integration
+                type: object
+                x-kubernetes-preserve-unknown-fields: true
               integration:
                 description: Integration is an optional integration used to specify
                   custom parameters
@@ -84,6 +89,21 @@ spec:
                     items:
                       type: string
                     type: array
+                  errorHandler:
+                    description: ErrorHandlerSpec respresents an integration error
+                      handler to be used as default at runtime
+                    properties:
+                      parameters:
+                        description: ErrorHandlerParameters is an unstructured object
+                          representing the configuration parameters available for
+                          an error handler
+                        type: object
+                        x-kubernetes-preserve-unknown-fields: true
+                      type:
+                        type: string
+                      uri:
+                        type: string
+                    type: object
                   flows:
                     items:
                       description: Flow is an unstructured object representing a Camel
diff --git a/deploy/olm-catalog/camel-k-dev/1.4.1-snapshot/camel.apache.org_kameletbindings.yaml b/deploy/olm-catalog/camel-k-dev/1.4.1-snapshot/camel.apache.org_kameletbindings.yaml
index c9a8b8724..60ac3ba49 100644
--- a/deploy/olm-catalog/camel-k-dev/1.4.1-snapshot/camel.apache.org_kameletbindings.yaml
+++ b/deploy/olm-catalog/camel-k-dev/1.4.1-snapshot/camel.apache.org_kameletbindings.yaml
@@ -63,6 +63,11 @@ spec:
           spec:
             description: KameletBindingSpec --
             properties:
+              errorHandler:
+                description: ErrorHandler is an optional handler called upon an error
+                  occuring in the integration
+                type: object
+                x-kubernetes-preserve-unknown-fields: true
               integration:
                 description: Integration is an optional integration used to specify
                   custom parameters
diff --git a/deploy/traits.yaml b/deploy/traits.yaml
index cec7545a9..be470679c 100755
--- a/deploy/traits.yaml
+++ b/deploy/traits.yaml
@@ -271,6 +271,22 @@ traits:
     type: bool
     description: Enables injection of `NAMESPACE` and `POD_NAME` environment variables
       (default `true`)
+- name: error-handler
+  platform: true
+  profiles:
+  - Kubernetes
+  - Knative
+  - OpenShift
+  description: The error-handler is a platform trait used to inject Error Handler
+    source into the integration runtime.
+  properties:
+  - name: enabled
+    type: bool
+    description: Can be used to enable or disable a trait. All traits share this common
+      property.
+  - name: ref
+    type: string
+    description: The error handler ref name provided or found in application properties
 - name: gc
   platform: false
   profiles:
diff --git a/docs/modules/ROOT/nav.adoc b/docs/modules/ROOT/nav.adoc
index 93b1e1e4c..126f7e12a 100644
--- a/docs/modules/ROOT/nav.adoc
+++ b/docs/modules/ROOT/nav.adoc
@@ -40,6 +40,7 @@
 ** xref:traits:deployer.adoc[Deployer]
 ** xref:traits:deployment.adoc[Deployment]
 ** xref:traits:environment.adoc[Environment]
+** xref:traits:error-handler.adoc[Error Handler]
 ** xref:traits:gc.adoc[Gc]
 ** xref:traits:ingress.adoc[Ingress]
 ** xref:traits:istio.adoc[Istio]
diff --git a/docs/modules/ROOT/pages/kamelets/kameletbindings-error-handler.adoc b/docs/modules/ROOT/pages/kamelets/kameletbindings-error-handler.adoc
new file mode 100644
index 000000000..fe3f1cef2
--- /dev/null
+++ b/docs/modules/ROOT/pages/kamelets/kameletbindings-error-handler.adoc
@@ -0,0 +1,156 @@
+[[kameletbindings-error-handler]]
+= Kamelet Bindings Error Handler
+
+`Kamelet Binding`s offer a mechanism to specify an error policy to adopt in case an event produced by a `source` or consumed by a `sink`. Through the definition of an `errorHandler` you will be able to apply certain logic to the failing event, such as simply logging, ignoring the event or posting the event to a `Dead Letter Channel`.
+
+[source,yaml]
+----
+apiVersion: camel.apache.org/v1alpha1
+kind: KameletBinding
+metadata:
+  name: my-kamelet-binding
+spec:
+  source: # <1>
+...
+  sink: # <2>
+...
+  errorHandler: # <3>
+----
+<1> Reference to the source that provides data
+<2> Reference to the sink where data should be sent to
+<3> Error Handler Configuration
+
+[[kameletbindings-error-handler-types]]
+== Error Handler Types
+
+We have different types of error handler: `ǹone`, `log`, `dead-letter-channel`, `bean`, `ref`. The `errorHandler` parameter is optional. The runtime will use the configuration to create an https://camel.apache.org/manual/latest/error-handler.html[Error Handler] as specified by Apache Camel.
+
+[[kameletbindings-error-handler-none]]
+=== No error handler
+
+There may be certain cases where you want to just ignore any failure happening on your integration. In this situation just use a `ǹone` error handler. Under the hood the configuration will create a https://camel.apache.org/manual/latest/error-handler.html#ErrorHandler-NoErrorHandler[`No` error handler builder]. 
+
+[source,yaml]
+----
+apiVersion: camel.apache.org/v1alpha1
+kind: KameletBinding
+metadata:
+  name: my-kamelet-binding
+spec:
+  source:
+...
+  sink: 
+...
+  errorHandler: 
+    none: # <1>
+----
+<1> `none` error handler does not expect any configuration
+
+[[kameletbindings-error-handler-log]]
+=== Log error handler
+
+Apache Camel offers a default behavior for handling any failure: log to standard output. However you can use the `log` error handler to specify other behaviors such as redelivery or delay policy. Under the hood the configuration will create a https://camel.apache.org/manual/latest/error-handler.html#_defaulterrorhandler[`Default` error handler builder].
+
+
+[source,yaml]
+----
+apiVersion: camel.apache.org/v1alpha1
+kind: KameletBinding
+metadata:
+  name: my-kamelet-binding
+spec:
+  source:
+...
+  sink: 
+...
+  errorHandler: 
+    log:
+      parameters: # <1>
+        maximumRedeliveries: 3
+        redeliveryDelay: 2000
+----
+<1> Parameters belonging to the `log` error handler type
+
+[[kameletbindings-error-handler-dlc]]
+=== Dead Letter Channel error handler
+
+The `Dead Letter Channel` is probably the most interesting error handler type as it allows you to redirect any failing event to any other component, such as a third party URI, a queue or even another `Kamelet` which will be performing certain logic with the failing event. Under the hood the configuration will create a https://camel.apache.org/manual/latest/error-handler.html#_dead_letter_channel[`Dead Letter Channel` error handler builder].
+
+[source,yaml]
+----
+apiVersion: camel.apache.org/v1alpha1
+kind: KameletBinding
+metadata:
+  name: my-kamelet-binding
+spec:
+  source:
+...
+  sink: 
+...
+  errorHandler: 
+    dead-letter-channel:
+      endpoint: 
+        ref: # <1>
+          kind: Kamelet
+          apiVersion: camel.apache.org/v1alpha1
+          name: error-handler
+        properties:
+          message: "ERROR!" # <2>
+          ...
+      parameters:
+        maximumRedeliveries: 1 # <3>
+        ... 
+----
+<1> You can use `ref` or `uri`. `ref` will be interpreted by the operator according the `kind`, `apiVersion` and `name`. You can use any `Kamelet`, `KafkaTopic` channel or `Knative` destination.
+<2> Properties belonging to the endpoint (in this example, to the `Kamelet` named error handler)
+<3> Parameters belonging to the `dead-letter-channel` error handler type
+
+[[kameletbindings-error-handler-bean]]
+=== Bean error handler
+
+With the `Bean` error handler you can extend the functionality of the Error Handler by providing a custom bean to be used as Error Handler.
+
+[source,yaml]
+----
+apiVersion: camel.apache.org/v1alpha1
+kind: KameletBinding
+metadata:
+  name: my-kamelet-binding
+spec:
+  source:
+...
+  sink: 
+...
+  errorHandler: 
+    bean:
+      type: "org.apache.camel.builder.DeadLetterChannelBuilder" # <1>
+      properties: # <2>
+        deadLetterUri: log:error
+        ... 
+----
+<1> Fully qualified name of the ErrorHandlerBuilder
+<2> Properties expected by your type
+
+[[kameletbindings-error-handler-ref]]
+=== Ref error handler
+
+With the `Ref` error handler you can use any `Bean` that is expected to be found in the Camel registry at runtime.
+
+[source,yaml]
+----
+apiVersion: camel.apache.org/v1alpha1
+kind: KameletBinding
+metadata:
+  name: my-kamelet-binding
+spec:
+  source:
+...
+  sink: 
+...
+  errorHandler: 
+    ref: my-custom-builder # <1>
+... 
+----
+<1> The name of the bean to be looked up at runtime
+
+NOTE: make sure to have the `ref` correctly bind at runtime.
\ No newline at end of file
diff --git a/docs/modules/ROOT/pages/kamelets/kamelets-user.adoc b/docs/modules/ROOT/pages/kamelets/kamelets-user.adoc
index 4a334cb4a..ef48761c5 100644
--- a/docs/modules/ROOT/pages/kamelets/kamelets-user.adoc
+++ b/docs/modules/ROOT/pages/kamelets/kamelets-user.adoc
@@ -356,6 +356,10 @@ This KameletBinding explicitly defines an URI where data is going to be pushed.
 NOTE: the `uri` option is also conventionally used in Knative to specify a non-kubernetes destination.
 To comply with the Knative specifications, in case an "http" or "https" URI is used, Camel will send https://cloudevents.io/[CloudEvents] to the destination.
 
+=== Error Handling
+
+You can configure an error handler in order to specify what to do when some event ends up with failure. See xref:kamelets/kameletbindings-error-handler.adoc[Kamelet Bindings Error Handler User Guide] for more detail.
+
 [[kamelets-specification]]
 == Kamelet Specification
 
diff --git a/docs/modules/ROOT/pages/kamelets/kamelets.adoc b/docs/modules/ROOT/pages/kamelets/kamelets.adoc
index 6ba3ff5cc..c0d0327b7 100644
--- a/docs/modules/ROOT/pages/kamelets/kamelets.adoc
+++ b/docs/modules/ROOT/pages/kamelets/kamelets.adoc
@@ -4,4 +4,5 @@
 This section of the documentation contains information about Kamelets from both a user and a developer point of view:
 
 - The xref:kamelets/kamelets-user.adoc[Kamelets User Guide] is for users that want to use Kamelets for connecting systems with their applications
+- The xref:kamelets/kameletbindings-error-handler.adoc[Kamelet Bindings Error Handler User Guide] explains how to configure an error handler on a `Kamelet Binding`
 - The xref:kamelets/kamelets-dev.adoc[Kamelets Developer Guide] is a reference for developers that want to create new Kamelets and share them with users
diff --git a/docs/modules/traits/pages/error-handler.adoc b/docs/modules/traits/pages/error-handler.adoc
new file mode 100755
index 000000000..7ddfd172c
--- /dev/null
+++ b/docs/modules/traits/pages/error-handler.adoc
@@ -0,0 +1,36 @@
+= Error Handler Trait
+
+// Start of autogenerated code - DO NOT EDIT! (description)
+The error-handler is a platform trait used to inject Error Handler source into the integration runtime.
+
+
+This trait is available in the following profiles: **Kubernetes, Knative, OpenShift**.
+
+WARNING: The error-handler trait is a *platform trait*: disabling it may compromise the platform functionality.
+
+// End of autogenerated code - DO NOT EDIT! (description)
+// Start of autogenerated code - DO NOT EDIT! (configuration)
+== Configuration
+
+Trait properties can be specified when running any integration with the CLI:
+[source,console]
+----
+$ kamel run --trait error-handler.[key]=[value] --trait error-handler.[key2]=[value2] integration.groovy
+----
+The following configuration options are available:
+
+[cols="2m,1m,5a"]
+|===
+|Property | Type | Description
+
+| error-handler.enabled
+| bool
+| Can be used to enable or disable a trait. All traits share this common property.
+
+| error-handler.ref
+| string
+| The error handler ref name provided or found in application properties
+
+|===
+
+// End of autogenerated code - DO NOT EDIT! (configuration)
diff --git a/examples/kafka/sasl_ssl/README.md b/examples/kafka/sasl_ssl/README.md
new file mode 100644
index 000000000..28251af0b
--- /dev/null
+++ b/examples/kafka/sasl_ssl/README.md
@@ -0,0 +1,23 @@
+# Kafka Camel K SASL SSL example
+
+This example shows how Camel K can be used to connect to a generic Kafka broker using SASL SSL authentication mechanism. Edit the `application` properties file with the proper values to be able to authenticate a Kafka Broker and use any Topic.
+
+For convenience create a secret to contain the sensitive properties in the `application.properties` file:
+
+```
+kubectl create secret generic kafka-props --from-file application.properties
+```
+
+Finally run this sample using the command:
+
+```
+kamel run --secret kafka-props SaslSSLKafkaConsumer.java --dev
+```
+
+A consumer will start logging the events found in the Topic:
+
+```
+[1] 2021-04-29 08:57:16,894 INFO  [FromKafka] (Camel (camel-1) thread #0 - KafkaConsumer[my-first-test]) Producing message #1096
+[1] 2021-04-29 08:57:18,995 INFO  [FromKafka] (Camel (camel-1) thread #0 - KafkaConsumer[my-first-test]) Producing message #1097
+[1] 2021-04-29 08:57:20,879 INFO  [FromKafka] (Camel (camel-1) thread #0 - KafkaConsumer[my-first-test]) Producing message #1098
+```
\ No newline at end of file
diff --git a/examples/kafka/sasl_ssl/SaslSSLKafkaConsumer.java b/examples/kafka/sasl_ssl/SaslSSLKafkaConsumer.java
new file mode 100644
index 000000000..6e14539ab
--- /dev/null
+++ b/examples/kafka/sasl_ssl/SaslSSLKafkaConsumer.java
@@ -0,0 +1,31 @@
+/*
+ * 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.
+ */
+
+// kubectl create secret generic kafka-props --from-file application.properties
+// kamel run --secret kafka-props SaslSSLKafkaConsumer.java --dev
+
+import org.apache.camel.builder.RouteBuilder;
+
+public class SaslSSLKafkaConsumer extends RouteBuilder {
+  @Override
+  public void configure() throws Exception {
+	log.info("About to start route: Kafka Server -> Log ");
+	from("kafka:{{consumer.topic}}")
+             .routeId("FromKafka")
+             .log("${body}");
+  }
+}
diff --git a/examples/kafka/sasl_ssl/application.properties b/examples/kafka/sasl_ssl/application.properties
new file mode 100644
index 000000000..533271da9
--- /dev/null
+++ b/examples/kafka/sasl_ssl/application.properties
@@ -0,0 +1,25 @@
+# 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.
+
+# Kafka config
+camel.component.kafka.brokers = *********************.kafka.my-cloud.service.com:443
+camel.component.kafka.security-protocol = SASL_SSL
+
+camel.component.kafka.sasl-mechanism = PLAIN
+camel.component.kafka.sasl-jaas-config=org.apache.kafka.common.security.plain.PlainLoginModule required username=*** password=***;
+
+consumer.topic = my-first-test
\ No newline at end of file
diff --git a/examples/kamelets/error-handler/error-handler.kamelet.yaml b/examples/kamelets/error-handler/error-handler.kamelet.yaml
new file mode 100644
index 000000000..04ce5573b
--- /dev/null
+++ b/examples/kamelets/error-handler/error-handler.kamelet.yaml
@@ -0,0 +1,73 @@
+# ---------------------------------------------------------------------------
+# 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/v1alpha1
+kind: Kamelet
+metadata:
+  name: error-handler
+spec:
+  definition:
+    title: "DLC and Log"
+    description: "Push an event to a kafka topic and log an error message"
+    required:
+      - kafka-brokers
+      - kafka-topic
+      - kafka-service-account-id
+      - kafka-service-account-secret
+      - log-message
+    properties:
+      kafka-brokers:
+        title: Kafka Brokers
+        description: the bootstrap server
+        type: string
+        example: "xyz-ins--rplssqfz-yyyyyyy-crojg.bf2.kafka.my-clud-service.com:443"    
+      kafka-topic:
+        title: Kafka Topic
+        description: the DLC topic
+        type: string
+        example: "my-dlc"    
+      kafka-service-account-id:
+        title: Service Account ID
+        description: the SA to use
+        type: string
+        example: "srvc-acct-xxxxxx-519b-453f-9f68-yyyyyyyyy"    
+      kafka-service-account-secret:
+        title: Service Account Secret
+        description: the SA secrete to use
+        type: string
+        example: "xxxxxxxxx-46c7-4c6c-a753-yyyyyyyyyyyyyyy"                
+      log-message:
+        title: Message
+        description: A message warning to log
+        type: string
+        example: "error while checking the source"    
+  flow:
+    from:
+      uri: kamelet:source
+      steps:
+      # First step: send to the DLC for future processing
+      - to:
+          uri: kafka:{{kafka-topic}}
+          parameters:
+            brokers: "{{kafka-brokers}}"
+            security-protocol: SASL_SSL
+            sasl-mechanism: PLAIN
+            sasl-jaas-config: "org.apache.kafka.common.security.plain.PlainLoginModule required username={{kafka-service-account-id}} password={{kafka-service-account-secret}};"
+      # Log an error message to notify about the failure
+      - set-body:
+          constant: "{{log-message}} - worry not, the event is stored in the DLC"
+      - to: "log:error-sink"  
diff --git a/examples/kamelets/error-handler/incremental-id-source.kamelet.yaml b/examples/kamelets/error-handler/incremental-id-source.kamelet.yaml
new file mode 100644
index 000000000..ee475da99
--- /dev/null
+++ b/examples/kamelets/error-handler/incremental-id-source.kamelet.yaml
@@ -0,0 +1,51 @@
+# ---------------------------------------------------------------------------
+# 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/v1alpha1
+kind: Kamelet
+metadata:
+  name: incremental-id-source
+  labels:
+    camel.apache.org/kamelet.type: "source"
+spec:
+  definition:
+    title: "Incremental ID Source"
+    description: "Produces periodic events with an incremental ID and fails when the ID contains any 0 character"
+    properties:
+      period:
+        title: Period
+        description: The interval between two events
+        type: integer
+        default: 1000
+  types:
+    out:
+      mediaType: text/plain
+  flow:
+    from:
+      uri: timer:tick
+      parameters:
+        period: "{{period}}"
+      steps:
+      - set-body:
+          simple: "Producing message #${exchangeProperty.CamelTimerCounter}"
+      - choice:
+          when:
+          - simple: "${bodyAs(String)} contains 0"
+            steps:
+            - set-body:
+                simple: "${mandatoryBodyAs(Boolean)}"
+      - to: "kamelet:sink"                
diff --git a/examples/kamelets/error-handler/kamelet-binding-error-handler.yaml b/examples/kamelets/error-handler/kamelet-binding-error-handler.yaml
new file mode 100644
index 000000000..3d3c3a54d
--- /dev/null
+++ b/examples/kamelets/error-handler/kamelet-binding-error-handler.yaml
@@ -0,0 +1,49 @@
+# ---------------------------------------------------------------------------
+# 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/v1alpha1
+kind: KameletBinding
+metadata:
+  name: kamelet-binding-error-handler
+spec:
+  source:
+    ref:
+      kind: Kamelet
+      apiVersion: camel.apache.org/v1alpha1
+      name: incremental-id-source
+  sink:
+    ref:
+      kind: Kamelet
+      apiVersion: camel.apache.org/v1alpha1
+      name: log-sink
+  errorHandler:
+    dead-letter-channel:
+      endpoint:
+        ref:
+          kind: Kamelet
+          apiVersion: camel.apache.org/v1alpha1
+          name: error-handler
+        properties:
+          log-message: "ERROR!"
+          kafka-brokers: ***
+          kafka-topic: my-first-test
+          kafka-service-account-id: ***
+          kafka-service-account-secret: ***
+      parameters:
+        maximumRedeliveries: 1
+        redeliveryDelay: 2000    
+    
diff --git a/examples/kamelets/error-handler/log-sink.kamelet.yaml b/examples/kamelets/error-handler/log-sink.kamelet.yaml
new file mode 100644
index 000000000..aa11481f4
--- /dev/null
+++ b/examples/kamelets/error-handler/log-sink.kamelet.yaml
@@ -0,0 +1,32 @@
+# ---------------------------------------------------------------------------
+# 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/v1alpha1
+kind: Kamelet
+metadata:
+  name: log-sink
+  labels:
+    camel.apache.org/kamelet.type: "sink"
+spec:
+  definition:
+    title: "Regular Log Sink"
+    description: "Consume events from a channel"  
+  flow:
+    from:
+      uri: kamelet:source
+      steps:
+      - to: "log:ok"
diff --git a/examples/kamelets/error-handler/readme.md b/examples/kamelets/error-handler/readme.md
new file mode 100644
index 000000000..abf8ad61f
--- /dev/null
+++ b/examples/kamelets/error-handler/readme.md
@@ -0,0 +1,127 @@
+# Kamelets Binding Error Handler example
+This example shows how to create a simple _source_ `Kamelet` which sends periodically events (and certain failures). The events are consumed by a log _sink_ in a `KameletBinding`. With the support of the `ErrorHandler` we will be able to redirect all errors to a `Dead Letter Channel` _error-handler_ `Kamelet` whose goal is to store the events in a `Kafka` topic and provide a nice log notifying us about the error happened.
+
+## Incremental ID Source Kamelet
+First of all, you must install the _incremental-id-source_ Kamelet defined in `incremental-id-source.kamelet.yaml` file. This source will emit events every second with an autoincrement counter that will be forced to fail when the number 0 is caught. With this trick, we will simulate possible event faults.
+```
+$ kubectl apply -f incremental-id-source.kamelet.yaml
+```
+You can check the newly created `kamelet` checking the list of kamelets available:
+```
+$ kubectl get kamelets
+
+NAME                            PHASE
+incremental-id-source           Ready
+```
+## Log Sink Kamelet
+Now it's the turn of installing the log-sink_ Kamelet defined in `log-sink.kamelet.yaml` file:
+```
+$ kubectl apply -f log-sink.kamelet.yaml
+```
+You can check the newly created `kamelet` checking the list of kamelets available:
+```
+$ kubectl get kamelets
+
+NAME                    PHASE
+log-sink                Ready
+incremental-id-source   Ready
+```
+## Error handler Kamelet
+We finally install an error handler as specified in `error-handler.kamelet.yaml` file. Let's have a look at how it is configured:
+
+```
+apiVersion: camel.apache.org/v1alpha1
+kind: Kamelet
+metadata:
+  name: error-handler
+spec:
+  definition:
+    ...
+    properties:
+      kafka-brokers:
+        ...  
+      kafka-topic:
+        ...
+      kafka-service-account-id:
+        ...  
+      kafka-service-account-secret:
+        ...              
+      log-message:
+        ...    
+  flow:
+    from:
+      uri: kamelet:source
+      steps:
+      # First step: send to the DLC for future processing
+      - to:
+          uri: kafka:{{kafka-topic}}
+          parameters:
+            brokers: "{{kafka-brokers}}"
+            security-protocol: SASL_SSL
+            sasl-mechanism: PLAIN
+            sasl-jaas-config: "org.apache.kafka.common.security.plain.PlainLoginModule required username={{kafka-service-account-id}} password={{kafka-service-account-secret}};"
+      # Log an error message to notify about the failure
+      - set-body:
+          constant: "{{log-message}} - worry not, the event is stored in the DLC"
+      - to: "log:error-sink"
+```
+
+We first send the errored event to a kafka topic, and then, we send a simple notification message to output, just to let the user know that some issue happened. Let's apply it:
+
+```
+$ kubectl apply -f error-handler.kamelet.yaml
+```
+You can check the newly created `kamelet` listing the kamelets available:
+```
+$ kubectl get kamelets
+
+NAME                    PHASE
+error-handler           Ready
+log-sink                Ready
+incremental-id-source   Ready
+```
+## Error Handler Kamelet Binding
+We can now create a `KameletBinding` which is started by the _incremental-id-source_ `Kamelet` and log events to _log-sink_ `Kamelet`. As this will sporadically fail, we can configure an _errorHandler_ with the _error-handler_ `Kamelet` as **Dead Letter Channel**. We want to configure also some redelivery policies (1 retry, with a 2000 milliseconds delay). We can declare it as in `kamelet-binding-error-handler.yaml` file:
+```
+...
+  errorHandler:
+    dead-letter-channel:
+      endpoint:
+        ref:
+          kind: Kamelet
+          apiVersion: camel.apache.org/v1alpha1
+          name: error-handler
+        properties:
+          message: "ERROR!"
+      parameters:
+        maximumRedeliveries: 1
+        redeliveryDelay: 2000
+```
+Execute the following command to start the `Integration`:
+```
+kubectl apply -f kamelet-binding-error-handler.yaml
+```
+As soon as the `Integration` starts, it will log the events on the `ok` log channel and errors on the `error` log channel:
+```
+[1] 2021-04-29 08:35:08,875 INFO  [sink] (Camel (camel-1) thread #0 - timer://tick) Exchange[ExchangePattern: InOnly, BodyType: String, Body: Producing message #49]
+[1] 2021-04-29 08:35:11,878 INFO  [sink] (Camel (camel-1) thread #0 - timer://tick) Exchange[ExchangePattern: InOnly, BodyType: String, Body: Producing message #51]
+[1] 2021-04-29 08:35:12,088 INFO  [error-sink] (Camel (camel-1) thread #9 - KafkaProducer[my-first-test]) Exchange[ExchangePattern: InOnly, BodyType: String, Body: ERROR! - worry not, the event is stored in the DLC]
+[1] 2021-04-29 08:35:12,877 INFO  [sink] (Camel (camel-1) thread #0 - timer://tick) Exchange[ExchangePattern: InOnly, BodyType: String, Body: Producing message #52]
+```
+
+### Recover the errors from DLC
+
+If you're curious to know what was going on in the DLC side, you can use the example you found in [kafka sasl ssl consumer](../kafka/sasl_ssl/):
+
+´´´
+kamel run --secret kafka-props SaslSSLKafkaConsumer.java --dev
+...
+[1] 2021-04-29 08:57:08,636 INFO  [org.apa.kaf.com.uti.AppInfoParser] (Camel (camel-1) thread #0 - KafkaConsumer[my-first-test]) Kafka commitId: 448719dc99a19793
+[1] 2021-04-29 08:57:08,636 INFO  [org.apa.kaf.com.uti.AppInfoParser] (Camel (camel-1) thread #0 - KafkaConsumer[my-first-test]) Kafka startTimeMs: 1619686628635
+[1] 2021-04-29 08:57:08,637 INFO  [org.apa.cam.com.kaf.KafkaConsumer] (Camel (camel-1) thread #0 - KafkaConsumer[my-first-test]) Subscribing my-first-test-Thread 0 to topic my-first-test
+...
+[1] 2021-04-29 08:35:02,894 INFO  [FromKafka] (Camel (camel-1) thread #0 - KafkaConsumer[my-first-test]) Producing message #40
+[1] 2021-04-29 08:35:12,995 INFO  [FromKafka] (Camel (camel-1) thread #0 - KafkaConsumer[my-first-test]) Producing message #50
+[1] 2021-04-29 08:35:22,879 INFO  [FromKafka] (Camel (camel-1) thread #0 - KafkaConsumer[my-first-test]) Producing message #60
+...
+´´´
\ No newline at end of file
diff --git a/helm/camel-k/crds/crd-kamelet-binding.yaml b/helm/camel-k/crds/crd-kamelet-binding.yaml
index c9a8b8724..60ac3ba49 100644
--- a/helm/camel-k/crds/crd-kamelet-binding.yaml
+++ b/helm/camel-k/crds/crd-kamelet-binding.yaml
@@ -63,6 +63,11 @@ spec:
           spec:
             description: KameletBindingSpec --
             properties:
+              errorHandler:
+                description: ErrorHandler is an optional handler called upon an error
+                  occuring in the integration
+                type: object
+                x-kubernetes-preserve-unknown-fields: true
               integration:
                 description: Integration is an optional integration used to specify
                   custom parameters
diff --git a/pkg/apis/camel/v1/common_types.go b/pkg/apis/camel/v1/common_types.go
index a94d6042d..e5b327b82 100644
--- a/pkg/apis/camel/v1/common_types.go
+++ b/pkg/apis/camel/v1/common_types.go
@@ -206,8 +206,9 @@ type SourceSpec struct {
 type SourceType string
 
 const (
-	SourceTypeDefault  SourceType = ""
-	SourceTypeTemplate SourceType = "template"
+	SourceTypeDefault      SourceType = ""
+	SourceTypeTemplate     SourceType = "template"
+	SourceTypeErrorHandler SourceType = "errorHandler"
 )
 
 // DataSpec --
diff --git a/pkg/apis/camel/v1/integration_types_support.go b/pkg/apis/camel/v1/integration_types_support.go
index 53705c90b..f7e505e43 100644
--- a/pkg/apis/camel/v1/integration_types_support.go
+++ b/pkg/apis/camel/v1/integration_types_support.go
@@ -131,6 +131,19 @@ func (in *IntegrationSpec) AddDependency(dependency string) {
 	in.Dependencies = append(in.Dependencies, newDep)
 }
 
+// GetConfigurationProperty returns a configuration property
+func (in *IntegrationSpec) GetConfigurationProperty(property string) string {
+	for _, confSpec := range in.Configuration {
+		if confSpec.Type == "property" && strings.HasPrefix(confSpec.Value, property) {
+			splitConf := strings.Split(confSpec.Value, "=")
+			if len(splitConf) > 1 {
+				return splitConf[1]
+			}
+		}
+	}
+	return ""
+}
+
 // AddOrReplaceGeneratedResources --
 func (in *IntegrationStatus) AddOrReplaceGeneratedResources(resources ...ResourceSpec) {
 	newResources := make([]ResourceSpec, 0)
diff --git a/pkg/apis/camel/v1alpha1/error_handler_types.go b/pkg/apis/camel/v1alpha1/error_handler_types.go
new file mode 100644
index 000000000..5019796ad
--- /dev/null
+++ b/pkg/apis/camel/v1alpha1/error_handler_types.go
@@ -0,0 +1,229 @@
+/*
+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 v1alpha1
+
+import (
+	"encoding/json"
+	"fmt"
+
+	v1 "github.com/apache/camel-k/pkg/apis/camel/v1"
+)
+
+// ErrorHandlerRefName --
+const ErrorHandlerRefName = "camel.k.errorHandler.ref"
+
+// ErrorHandlerRefDefaultName --
+const ErrorHandlerRefDefaultName = "defaultErrorHandler"
+
+// ErrorHandlerAppPropertiesPrefix --
+const ErrorHandlerAppPropertiesPrefix = "camel.beans.defaultErrorHandler"
+
+// ErrorHandlerSpec represents an unstructured object for an error handler
+type ErrorHandlerSpec struct {
+	v1.RawMessage `json:",omitempty"`
+}
+
+// ErrorHandlerParameters represent an unstructured object for error handler parameters
+type ErrorHandlerParameters struct {
+	v1.RawMessage `json:",omitempty"`
+}
+
+// BeanProperties represent an unstructured object properties to be set on a bean
+type BeanProperties struct {
+	v1.RawMessage `json:",omitempty"`
+}
+
+// ErrorHandler is a generic interface that represent any type of error handler specification
+type ErrorHandler interface {
+	Type() ErrorHandlerType
+	Endpoint() *Endpoint
+	Configuration() (map[string]interface{}, error)
+}
+
+type baseErrorHandler struct {
+}
+
+// Type --
+func (e baseErrorHandler) Type() ErrorHandlerType {
+	return errorHandlerTypeBase
+}
+
+// Endpoint --
+func (e baseErrorHandler) Endpoint() *Endpoint {
+	return nil
+}
+
+// Configuration --
+func (e baseErrorHandler) Configuration() (map[string]interface{}, error) {
+	return nil, nil
+}
+
+// ErrorHandlerNone --
+type ErrorHandlerNone struct {
+	baseErrorHandler
+}
+
+// Type --
+func (e ErrorHandlerNone) Type() ErrorHandlerType {
+	return ErrorHandlerTypeNone
+}
+
+// Configuration --
+func (e ErrorHandlerNone) Configuration() (map[string]interface{}, error) {
+	return map[string]interface{}{
+		ErrorHandlerAppPropertiesPrefix: "#class:org.apache.camel.builder.NoErrorHandlerBuilder",
+		ErrorHandlerRefName:             ErrorHandlerRefDefaultName,
+	}, nil
+}
+
+// ErrorHandlerLog represent a default (log) error handler type
+type ErrorHandlerLog struct {
+	ErrorHandlerNone
+	Parameters *ErrorHandlerParameters `json:"parameters,omitempty"`
+}
+
+// Type --
+func (e ErrorHandlerLog) Type() ErrorHandlerType {
+	return ErrorHandlerTypeLog
+}
+
+// Configuration --
+func (e ErrorHandlerLog) Configuration() (map[string]interface{}, error) {
+	properties, err := e.ErrorHandlerNone.Configuration()
+	if err != nil {
+		return nil, err
+	}
+	properties[ErrorHandlerAppPropertiesPrefix] = "#class:org.apache.camel.builder.DefaultErrorHandlerBuilder"
+
+	if e.Parameters != nil {
+		var parameters map[string]interface{}
+		err := json.Unmarshal(e.Parameters.RawMessage, &parameters)
+		if err != nil {
+			return nil, err
+		}
+		for key, value := range parameters {
+			properties[ErrorHandlerAppPropertiesPrefix+"."+key] = value
+		}
+	}
+
+	return properties, nil
+}
+
+// ErrorHandlerDeadLetterChannel represents a dead letter channel error handler type
+type ErrorHandlerDeadLetterChannel struct {
+	ErrorHandlerLog
+	DLCEndpoint *Endpoint `json:"endpoint,omitempty"`
+}
+
+// Type --
+func (e ErrorHandlerDeadLetterChannel) Type() ErrorHandlerType {
+	return ErrorHandlerTypeDeadLetterChannel
+}
+
+// Endpoint --
+func (e ErrorHandlerDeadLetterChannel) Endpoint() *Endpoint {
+	return e.DLCEndpoint
+}
+
+// Configuration --
+func (e ErrorHandlerDeadLetterChannel) Configuration() (map[string]interface{}, error) {
+	properties, err := e.ErrorHandlerLog.Configuration()
+	if err != nil {
+		return nil, err
+	}
+	properties[ErrorHandlerAppPropertiesPrefix] = "#class:org.apache.camel.builder.DeadLetterChannelBuilder"
+
+	return properties, err
+}
+
+// ErrorHandlerRef represents a reference to an error handler builder available in the registry
+type ErrorHandlerRef struct {
+	baseErrorHandler
+	v1.RawMessage
+}
+
+// Type --
+func (e ErrorHandlerRef) Type() ErrorHandlerType {
+	return ErrorHandlerTypeRef
+}
+
+// Configuration --
+func (e ErrorHandlerRef) Configuration() (map[string]interface{}, error) {
+	var refName string
+	err := json.Unmarshal(e.RawMessage, &refName)
+	if err != nil {
+		return nil, err
+	}
+
+	properties := map[string]interface{}{
+		ErrorHandlerRefName: refName,
+	}
+
+	return properties, nil
+}
+
+// ErrorHandlerBean represents a bean error handler type
+type ErrorHandlerBean struct {
+	ErrorHandlerNone
+	BeanType       *string         `json:"type,omitempty"`
+	BeanProperties *BeanProperties `json:"properties,omitempty"`
+}
+
+// Type --
+func (e ErrorHandlerBean) Type() ErrorHandlerType {
+	return ErrorHandlerTypeBean
+}
+
+// Configuration --
+func (e ErrorHandlerBean) Configuration() (map[string]interface{}, error) {
+	properties, err := e.ErrorHandlerNone.Configuration()
+	if err != nil {
+		return nil, err
+	}
+	properties[ErrorHandlerAppPropertiesPrefix] = fmt.Sprintf("#class:%v", *e.BeanType)
+
+	if e.BeanProperties != nil {
+		var beanProperties map[string]interface{}
+		err := json.Unmarshal(e.BeanProperties.RawMessage, &beanProperties)
+		if err != nil {
+			return nil, err
+		}
+		for key, value := range beanProperties {
+			properties[ErrorHandlerAppPropertiesPrefix+"."+key] = value
+		}
+	}
+
+	return properties, err
+}
+
+// ErrorHandlerType --
+type ErrorHandlerType string
+
+const (
+	errorHandlerTypeBase ErrorHandlerType = ""
+	// ErrorHandlerTypeNone --
+	ErrorHandlerTypeNone ErrorHandlerType = "none"
+	// ErrorHandlerTypeLog --
+	ErrorHandlerTypeLog ErrorHandlerType = "log"
+	// ErrorHandlerTypeDeadLetterChannel --
+	ErrorHandlerTypeDeadLetterChannel ErrorHandlerType = "dead-letter-channel"
+	// ErrorHandlerTypeRef --
+	ErrorHandlerTypeRef ErrorHandlerType = "ref"
+	// ErrorHandlerTypeBean --
+	ErrorHandlerTypeBean ErrorHandlerType = "bean"
+)
diff --git a/pkg/apis/camel/v1alpha1/kamelet_binding_types.go b/pkg/apis/camel/v1alpha1/kamelet_binding_types.go
index 8fea2ac1b..cfe8779d1 100644
--- a/pkg/apis/camel/v1alpha1/kamelet_binding_types.go
+++ b/pkg/apis/camel/v1alpha1/kamelet_binding_types.go
@@ -32,6 +32,8 @@ type KameletBindingSpec struct {
 	Source Endpoint `json:"source,omitempty"`
 	// Sink is the destination of the integration defined by this binding
 	Sink Endpoint `json:"sink,omitempty"`
+	// ErrorHandler is an optional handler called upon an error occuring in the integration
+	ErrorHandler ErrorHandlerSpec `json:"errorHandler,omitempty"`
 	// Steps contains an optional list of intermediate steps that are executed between the Source and the Sink
 	Steps []Endpoint `json:"steps,omitempty"`
 }
@@ -51,9 +53,10 @@ type Endpoint struct {
 type EndpointType string
 
 const (
-	EndpointTypeSource EndpointType = "source"
-	EndpointTypeAction EndpointType = "action"
-	EndpointTypeSink   EndpointType = "sink"
+	EndpointTypeSource       EndpointType = "source"
+	EndpointTypeAction       EndpointType = "action"
+	EndpointTypeSink         EndpointType = "sink"
+	EndpointTypeErrorHandler EndpointType = "errorHandler"
 )
 
 // EndpointProperties is a key/value struct represented as JSON raw to allow numeric/boolean values
diff --git a/pkg/controller/kameletbinding/common.go b/pkg/controller/kameletbinding/common.go
index 411bde8d2..5da336ad6 100644
--- a/pkg/controller/kameletbinding/common.go
+++ b/pkg/controller/kameletbinding/common.go
@@ -53,6 +53,7 @@ func createIntegrationFor(ctx context.Context, c client.Client, kameletbinding *
 			},
 		},
 	}
+
 	// start from the integration spec defined in the binding
 	if kameletbinding.Spec.Integration != nil {
 		it.Spec = *kameletbinding.Spec.Integration.DeepCopy()
@@ -79,6 +80,11 @@ func createIntegrationFor(ctx context.Context, c client.Client, kameletbinding *
 	if err != nil {
 		return nil, errors.Wrap(err, "could not determine sink URI")
 	}
+	// error handler is optional
+	errorHandler, err := maybeErrorHandler(kameletbinding.Spec.ErrorHandler, bindingContext)
+	if err != nil {
+		return nil, errors.Wrap(err, "could not determine error handler")
+	}
 
 	steps := make([]*bindings.Binding, 0, len(kameletbinding.Spec.Steps))
 	for idx, step := range kameletbinding.Spec.Steps {
@@ -93,10 +99,13 @@ func createIntegrationFor(ctx context.Context, c client.Client, kameletbinding *
 		steps = append(steps, stepBinding)
 	}
 
-	allBindings := make([]*bindings.Binding, 0, len(steps)+2)
+	allBindings := make([]*bindings.Binding, 0, len(steps)+3)
 	allBindings = append(allBindings, from)
 	allBindings = append(allBindings, steps...)
 	allBindings = append(allBindings, to)
+	if errorHandler != nil {
+		allBindings = append(allBindings, errorHandler)
+	}
 
 	propList := make([]string, 0)
 	for _, b := range allBindings {
@@ -129,17 +138,17 @@ func createIntegrationFor(ctx context.Context, c client.Client, kameletbinding *
 		"to": to.URI,
 	})
 
-	flow := map[string]interface{}{
+	flowFrom := map[string]interface{}{
 		"from": map[string]interface{}{
 			"uri":   from.URI,
 			"steps": dslSteps,
 		},
 	}
-	encodedFlow, err := json.Marshal(flow)
+	encodedFrom, err := json.Marshal(flowFrom)
 	if err != nil {
 		return nil, err
 	}
-	it.Spec.Flows = append(it.Spec.Flows, v1.Flow{RawMessage: encodedFlow})
+	it.Spec.Flows = append(it.Spec.Flows, v1.Flow{RawMessage: encodedFrom})
 
 	return &it, nil
 }
diff --git a/pkg/controller/kameletbinding/error_handler.go b/pkg/controller/kameletbinding/error_handler.go
new file mode 100644
index 000000000..5325324f0
--- /dev/null
+++ b/pkg/controller/kameletbinding/error_handler.go
@@ -0,0 +1,110 @@
+/*
+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 kameletbinding
+
+import (
+	"encoding/json"
+	"fmt"
+
+	v1 "github.com/apache/camel-k/pkg/apis/camel/v1"
+	"github.com/apache/camel-k/pkg/apis/camel/v1alpha1"
+	"github.com/apache/camel-k/pkg/util/bindings"
+	"github.com/pkg/errors"
+)
+
+func maybeErrorHandler(errHandlConf v1alpha1.ErrorHandlerSpec, bindingContext bindings.BindingContext) (*bindings.Binding, error) {
+	var errorHandlerBinding *bindings.Binding
+	if errHandlConf.RawMessage != nil {
+		errorHandlerSpec, err := parseErrorHandler(errHandlConf.RawMessage)
+		if err != nil {
+			return nil, errors.Wrap(err, "could not parse error handler")
+		}
+		// We need to get the translated URI from any referenced resource (ie, kamelets)
+		if errorHandlerSpec.Type() == v1alpha1.ErrorHandlerTypeDeadLetterChannel {
+			errorHandlerBinding, err = bindings.Translate(bindingContext, bindings.EndpointContext{Type: v1alpha1.EndpointTypeErrorHandler}, *errorHandlerSpec.Endpoint())
+			if err != nil {
+				return nil, errors.Wrap(err, "could not determine error handler URI")
+			}
+		} else {
+			// Create a new binding otherwise in order to store error handler application properties
+			errorHandlerBinding = &bindings.Binding{
+				ApplicationProperties: make(map[string]string),
+			}
+		}
+
+		err = setErrorHandlerConfiguration(errorHandlerBinding, errorHandlerSpec)
+		if err != nil {
+			return nil, errors.Wrap(err, "could not set integration error handler")
+		}
+
+		return errorHandlerBinding, nil
+	}
+	return nil, nil
+}
+
+func parseErrorHandler(rawMessage v1.RawMessage) (v1alpha1.ErrorHandler, error) {
+	var properties map[v1alpha1.ErrorHandlerType]v1.RawMessage
+	err := json.Unmarshal(rawMessage, &properties)
+	if err != nil {
+		return nil, err
+	}
+	if len(properties) > 1 {
+		return nil, errors.Errorf("You must provide just 1 error handler, provided %d", len(properties))
+	}
+
+	for errHandlType, errHandlValue := range properties {
+		var dst v1alpha1.ErrorHandler
+		switch errHandlType {
+		case v1alpha1.ErrorHandlerTypeNone:
+			dst = new(v1alpha1.ErrorHandlerNone)
+		case v1alpha1.ErrorHandlerTypeLog:
+			dst = new(v1alpha1.ErrorHandlerLog)
+		case v1alpha1.ErrorHandlerTypeDeadLetterChannel:
+			dst = new(v1alpha1.ErrorHandlerDeadLetterChannel)
+		case v1alpha1.ErrorHandlerTypeRef:
+			dst = new(v1alpha1.ErrorHandlerRef)
+		case v1alpha1.ErrorHandlerTypeBean:
+			dst = new(v1alpha1.ErrorHandlerBean)
+		default:
+			return nil, errors.Errorf("Unknown error handler type %s", errHandlType)
+		}
+
+		err := json.Unmarshal(errHandlValue, dst)
+		if err != nil {
+			return nil, err
+		}
+
+		return dst, nil
+	}
+
+	return nil, errors.New("You must provide any supported error handler")
+}
+
+func setErrorHandlerConfiguration(errorHandlerBinding *bindings.Binding, errorHandler v1alpha1.ErrorHandler) error {
+	properties, err := errorHandler.Configuration()
+	if err != nil {
+		return err
+	}
+	for key, value := range properties {
+		errorHandlerBinding.ApplicationProperties[key] = fmt.Sprintf("%v", value)
+	}
+	if errorHandler.Type() == v1alpha1.ErrorHandlerTypeDeadLetterChannel && errorHandlerBinding.URI != "" {
+		errorHandlerBinding.ApplicationProperties[fmt.Sprintf("%s.deadLetterUri", v1alpha1.ErrorHandlerAppPropertiesPrefix)] = fmt.Sprintf("%v", errorHandlerBinding.URI)
+	}
+	return nil
+}
diff --git a/pkg/trait/error_handler.go b/pkg/trait/error_handler.go
new file mode 100644
index 000000000..aa59b4e51
--- /dev/null
+++ b/pkg/trait/error_handler.go
@@ -0,0 +1,91 @@
+/*
+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 trait
+
+import (
+	v1 "github.com/apache/camel-k/pkg/apis/camel/v1"
+	"github.com/apache/camel-k/pkg/apis/camel/v1alpha1"
+	"gopkg.in/yaml.v2"
+)
+
+// The error-handler is a platform trait used to inject Error Handler source into the integration runtime.
+//
+// +camel-k:trait=error-handler
+type errorHandlerTrait struct {
+	BaseTrait `property:",squash"`
+	// The error handler ref name provided or found in application properties
+	ErrorHandlerRef string `property:"ref" json:"ref,omitempty"`
+}
+
+func newErrorHandlerTrait() Trait {
+	return &errorHandlerTrait{
+		BaseTrait: NewBaseTrait("error-handler", 500),
+	}
+}
+
+// IsPlatformTrait overrides base class method
+func (t *errorHandlerTrait) IsPlatformTrait() bool {
+	return true
+}
+
+func (t *errorHandlerTrait) Configure(e *Environment) (bool, error) {
+	if t.Enabled != nil && !*t.Enabled {
+		return false, nil
+	}
+
+	if !e.IntegrationInPhase(v1.IntegrationPhaseInitialization, v1.IntegrationPhaseDeploying, v1.IntegrationPhaseRunning) {
+		return false, nil
+	}
+
+	if t.ErrorHandlerRef == "" {
+		t.ErrorHandlerRef = e.Integration.Spec.GetConfigurationProperty(v1alpha1.ErrorHandlerRefName)
+	}
+
+	return t.ErrorHandlerRef != "", nil
+}
+
+func (t *errorHandlerTrait) Apply(e *Environment) error {
+	if e.IntegrationInPhase(v1.IntegrationPhaseInitialization) {
+		return t.addErrorHandlerAsSource(e)
+	}
+	return nil
+}
+
+func (t *errorHandlerTrait) addErrorHandlerAsSource(e *Environment) error {
+	flowErrorHandler := map[string]interface{}{
+		"error-handler": map[string]string{
+			"ref": t.ErrorHandlerRef,
+		},
+	}
+	encodedFlowErrorHandler, err := yaml.Marshal([]map[string]interface{}{flowErrorHandler})
+	if err != nil {
+		return err
+	}
+	errorHandlerSource := v1.SourceSpec{
+		DataSpec: v1.DataSpec{
+			Name:    "camel-k-embedded-error-handler.yaml",
+			Content: string(encodedFlowErrorHandler),
+		},
+		Language: v1.LanguageYaml,
+		Type:     v1.SourceTypeErrorHandler,
+	}
+
+	e.Integration.Status.AddOrReplaceGeneratedSources(errorHandlerSource)
+
+	return nil
+}
diff --git a/pkg/trait/kamelets.go b/pkg/trait/kamelets.go
index df437d9fe..916b4b280 100644
--- a/pkg/trait/kamelets.go
+++ b/pkg/trait/kamelets.go
@@ -94,8 +94,8 @@ func (t *kameletsTrait) Configure(e *Environment) (bool, error) {
 	}
 
 	if t.Auto == nil || *t.Auto {
+		var kamelets []string
 		if t.List == "" {
-			var kamelets []string
 			sources, err := kubernetes.ResolveIntegrationSources(e.C, e.Client, e.Integration, e.Resources)
 			if err != nil {
 				return false, err
@@ -105,17 +105,25 @@ func (t *kameletsTrait) Configure(e *Environment) (bool, error) {
 				util.StringSliceUniqueConcat(&kamelets, extractKamelets(meta.ToURIs))
 				return true
 			})
+		}
+		// Check if a Kamelet is configured as default error handler URI
+		defaultErrorHandlerURI := e.Integration.Spec.GetConfigurationProperty(v1alpha1.ErrorHandlerAppPropertiesPrefix + ".deadLetterUri")
+		if defaultErrorHandlerURI != "" {
+			if strings.HasPrefix(defaultErrorHandlerURI, "kamelet:") {
+				kamelets = append(kamelets, extractKamelet(defaultErrorHandlerURI))
+			}
+		}
+
+		if len(kamelets) > 0 {
 			sort.Strings(kamelets)
 			t.List = strings.Join(kamelets, ",")
 		}
-
 	}
 
 	return len(t.getKameletKeys()) > 0, nil
 }
 
 func (t *kameletsTrait) Apply(e *Environment) error {
-
 	if e.IntegrationInPhase(v1.IntegrationPhaseInitialization, v1.IntegrationPhaseRunning) {
 		if err := t.addKamelets(e); err != nil {
 			return err
@@ -404,10 +412,18 @@ func integrationSourceFromKameletSource(e *Environment, kamelet *v1alpha1.Kamele
 
 func extractKamelets(uris []string) (kamelets []string) {
 	for _, uri := range uris {
-		matches := kameletNameRegexp.FindStringSubmatch(uri)
-		if len(matches) > 1 {
-			kamelets = append(kamelets, matches[1])
+		kamelet := extractKamelet(uri)
+		if kamelet != "" {
+			kamelets = append(kamelets, kamelet)
 		}
 	}
 	return
 }
+
+func extractKamelet(uri string) (kamelet string) {
+	matches := kameletNameRegexp.FindStringSubmatch(uri)
+	if len(matches) > 1 {
+		return matches[1]
+	}
+	return ""
+}
diff --git a/pkg/trait/trait_register.go b/pkg/trait/trait_register.go
index a43a2d71f..f8e079459 100644
--- a/pkg/trait/trait_register.go
+++ b/pkg/trait/trait_register.go
@@ -26,6 +26,7 @@ func init() {
 	AddToTraits(newOpenAPITrait)
 	AddToTraits(newKnativeTrait)
 	AddToTraits(newKameletsTrait)
+	AddToTraits(newErrorHandlerTrait)
 	AddToTraits(newDependenciesTrait)
 	AddToTraits(newBuilderTrait)
 	AddToTraits(newQuarkusTrait)


Test patch:
diff --git a/e2e/common/traits/error_handler_test.go b/e2e/common/traits/error_handler_test.go
new file mode 100644
index 000000000..506eba4e6
--- /dev/null
+++ b/e2e/common/traits/error_handler_test.go
@@ -0,0 +1,57 @@
+// +build integration
+
+// To enable compilation of this file in Goland, go to "Settings -> Go -> Vendoring & Build Tags -> Custom Tags" and add "integration"
+
+/*
+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 traits
+
+import (
+	"testing"
+
+	. "github.com/onsi/gomega"
+
+	v1 "k8s.io/api/core/v1"
+
+	. "github.com/apache/camel-k/e2e/support"
+	camelv1 "github.com/apache/camel-k/pkg/apis/camel/v1"
+)
+
+func TestErrorHandlerTrait(t *testing.T) {
+	WithNewTestNamespace(t, func(ns string) {
+		Expect(Kamel("install", "-n", ns).Execute()).To(Succeed())
+
+		t.Run("Run errored integration with error handler", func(t *testing.T) {
+			name := "error-handler"
+			Expect(Kamel("run", "-n", ns, "files/ErroredRoute.java",
+				"--name", name,
+				"-t", "error-handler.enabled=true",
+				"-t", "error-handler.ref=defaultErrorHandler",
+				"-p", "camel.beans.defaultErrorHandler=#class:org.apache.camel.builder.DeadLetterChannelBuilder",
+				"-p", "camel.beans.defaultErrorHandler.deadLetterUri=log:my-special-error-handler-in-place?level=ERROR&showCaughtException=false&showBody=false&showBodyType=false&showExchangePattern=false",
+			).Execute()).To(Succeed())
+			Eventually(IntegrationPodPhase(ns, name), TestTimeoutMedium).Should(Equal(v1.PodRunning))
+			Eventually(IntegrationCondition(ns, name, camelv1.IntegrationConditionReady), TestTimeoutShort).Should(Equal(v1.ConditionTrue))
+			Eventually(IntegrationLogs(ns, name), TestTimeoutShort).ShouldNot(ContainSubstring("InvalidPayloadException"))
+			Eventually(IntegrationLogs(ns, name), TestTimeoutShort).Should(ContainSubstring("my-special-error-handler-in-place"))
+		})
+
+		// Clean up
+		Expect(Kamel("delete", "--all", "-n", ns).Execute()).To(Succeed())
+	})
+}
diff --git a/e2e/common/traits/files/ErroredRoute.java b/e2e/common/traits/files/ErroredRoute.java
new file mode 100644
index 000000000..702e16372
--- /dev/null
+++ b/e2e/common/traits/files/ErroredRoute.java
@@ -0,0 +1,29 @@
+/*
+ * 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.
+ */
+
+import org.apache.camel.builder.RouteBuilder;
+
+public class ErroredRoute extends RouteBuilder {
+  @Override
+  public void configure() throws Exception {
+	  from("timer:tick")
+    .setBody().simple("Hello")
+    // We want to force a failure
+	  .setBody().simple("${mandatoryBodyAs(Boolean)}")
+    .to("log:info");
+  }
+}
diff --git a/pkg/controller/kameletbinding/error_handler_test.go b/pkg/controller/kameletbinding/error_handler_test.go
new file mode 100644
index 000000000..94e6643ba
--- /dev/null
+++ b/pkg/controller/kameletbinding/error_handler_test.go
@@ -0,0 +1,134 @@
+/*
+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 kameletbinding
+
+import (
+	"fmt"
+	"testing"
+
+	"github.com/apache/camel-k/pkg/apis/camel/v1alpha1"
+	"github.com/stretchr/testify/assert"
+)
+
+func TestParseErrorHandlerNoneDoesSucceed(t *testing.T) {
+	noErrorHandler, err := parseErrorHandler(
+		[]byte(`{"none": null}`),
+	)
+	assert.Nil(t, err)
+	assert.Equal(t, v1alpha1.ErrorHandlerTypeNone, noErrorHandler.Type())
+	parameters, err := noErrorHandler.Configuration()
+	assert.Nil(t, err)
+	assert.Equal(t, "#class:org.apache.camel.builder.NoErrorHandlerBuilder", parameters[v1alpha1.ErrorHandlerAppPropertiesPrefix])
+	assert.Equal(t, v1alpha1.ErrorHandlerRefDefaultName, parameters[v1alpha1.ErrorHandlerRefName])
+}
+
+func TestParseErrorHandlerLogDoesSucceed(t *testing.T) {
+	logErrorHandler, err := parseErrorHandler(
+		[]byte(`{"log": null}`),
+	)
+	assert.Nil(t, err)
+	assert.Equal(t, v1alpha1.ErrorHandlerTypeLog, logErrorHandler.Type())
+	parameters, err := logErrorHandler.Configuration()
+	assert.Nil(t, err)
+	assert.Equal(t, "#class:org.apache.camel.builder.DefaultErrorHandlerBuilder", parameters[v1alpha1.ErrorHandlerAppPropertiesPrefix])
+	assert.Equal(t, v1alpha1.ErrorHandlerRefDefaultName, parameters[v1alpha1.ErrorHandlerRefName])
+}
+
+func TestParseErrorHandlerLogWithParametersDoesSucceed(t *testing.T) {
+	logErrorHandler, err := parseErrorHandler(
+		[]byte(`{"log": {"parameters": {"param1": "value1", "param2": "value2"}}}`),
+	)
+	assert.Nil(t, err)
+	assert.Equal(t, v1alpha1.ErrorHandlerTypeLog, logErrorHandler.Type())
+	parameters, err := logErrorHandler.Configuration()
+	assert.Nil(t, err)
+	assert.Equal(t, "#class:org.apache.camel.builder.DefaultErrorHandlerBuilder", parameters[v1alpha1.ErrorHandlerAppPropertiesPrefix])
+	assert.Equal(t, "value1", parameters["camel.beans.defaultErrorHandler.param1"])
+	assert.Equal(t, "value2", parameters["camel.beans.defaultErrorHandler.param2"])
+	assert.Equal(t, v1alpha1.ErrorHandlerRefDefaultName, parameters[v1alpha1.ErrorHandlerRefName])
+}
+
+func TestParseErrorHandlerDLCDoesSucceed(t *testing.T) {
+	fmt.Println("Test")
+	dlcErrorHandler, err := parseErrorHandler(
+		[]byte(`{"dead-letter-channel": {"endpoint": {"uri": "someUri"}}}`),
+	)
+	assert.Nil(t, err)
+	assert.NotNil(t, dlcErrorHandler)
+	assert.Equal(t, v1alpha1.ErrorHandlerTypeDeadLetterChannel, dlcErrorHandler.Type())
+	assert.Equal(t, "someUri", *dlcErrorHandler.Endpoint().URI)
+	parameters, err := dlcErrorHandler.Configuration()
+	assert.Nil(t, err)
+	assert.Equal(t, "#class:org.apache.camel.builder.DeadLetterChannelBuilder", parameters[v1alpha1.ErrorHandlerAppPropertiesPrefix])
+	assert.Equal(t, v1alpha1.ErrorHandlerRefDefaultName, parameters[v1alpha1.ErrorHandlerRefName])
+}
+
+func TestParseErrorHandlerDLCWithParametersDoesSucceed(t *testing.T) {
+	dlcErrorHandler, err := parseErrorHandler(
+		[]byte(`{
+			"dead-letter-channel": {
+				"endpoint": {
+					"uri": "someUri"
+					}, 
+				"parameters": 
+					{"param1": "value1", "param2": "value2"}
+			}
+		}`),
+	)
+	assert.Nil(t, err)
+	assert.NotNil(t, dlcErrorHandler)
+	assert.Equal(t, v1alpha1.ErrorHandlerTypeDeadLetterChannel, dlcErrorHandler.Type())
+	assert.Equal(t, "someUri", *dlcErrorHandler.Endpoint().URI)
+	parameters, err := dlcErrorHandler.Configuration()
+	assert.Nil(t, err)
+	assert.Equal(t, "#class:org.apache.camel.builder.DeadLetterChannelBuilder", parameters[v1alpha1.ErrorHandlerAppPropertiesPrefix])
+	assert.Equal(t, v1alpha1.ErrorHandlerRefDefaultName, parameters[v1alpha1.ErrorHandlerRefName])
+	assert.Equal(t, "value1", parameters["camel.beans.defaultErrorHandler.param1"])
+	assert.Equal(t, "value2", parameters["camel.beans.defaultErrorHandler.param2"])
+}
+
+func TestParseErrorHandlerBeanWithParamsDoesSucceed(t *testing.T) {
+	beanErrorHandler, err := parseErrorHandler(
+		[]byte(`{
+			"bean": {
+				"type": "com.acme.MyType",
+				"properties": 
+					{"beanProp1": "value1", "beanProp2": "value2"}
+			}
+		}`),
+	)
+	assert.Nil(t, err)
+	assert.Equal(t, v1alpha1.ErrorHandlerTypeBean, beanErrorHandler.Type())
+	parameters, err := beanErrorHandler.Configuration()
+	assert.Nil(t, err)
+	assert.Equal(t, "#class:com.acme.MyType", parameters[v1alpha1.ErrorHandlerAppPropertiesPrefix])
+	assert.Equal(t, v1alpha1.ErrorHandlerRefDefaultName, parameters[v1alpha1.ErrorHandlerRefName])
+	assert.Equal(t, "value1", parameters["camel.beans.defaultErrorHandler.beanProp1"])
+	assert.Equal(t, "value2", parameters["camel.beans.defaultErrorHandler.beanProp2"])
+}
+
+func TestParseErrorHandlerRefDoesSucceed(t *testing.T) {
+	refErrorHandler, err := parseErrorHandler(
+		[]byte(`{"ref": "my-registry-ref"}`),
+	)
+	assert.Nil(t, err)
+	assert.Equal(t, v1alpha1.ErrorHandlerTypeRef, refErrorHandler.Type())
+	parameters, err := refErrorHandler.Configuration()
+	assert.Nil(t, err)
+	assert.Equal(t, "my-registry-ref", parameters[v1alpha1.ErrorHandlerRefName])
+}
diff --git a/pkg/trait/trait_test.go b/pkg/trait/trait_test.go
index f8f75af80..e1bea3b04 100644
--- a/pkg/trait/trait_test.go
+++ b/pkg/trait/trait_test.go
@@ -507,7 +507,7 @@ func TestOnlySomeTraitsInfluenceBuild(t *testing.T) {
 
 func TestOnlySomeTraitsArePlatform(t *testing.T) {
 	c := NewTraitTestCatalog()
-	platformTraits := []string{"builder", "camel", "jvm", "container", "dependencies", "deployer", "deployment", "environment", "kamelets", "openapi", "owner", "platform", "quarkus"}
+	platformTraits := []string{"builder", "camel", "jvm", "container", "dependencies", "deployer", "deployment", "environment", "error-handler", "kamelets", "openapi", "owner", "platform", "quarkus"}
 
 	for _, trait := range c.allTraits() {
 		if trait.IsPlatformTrait() {