搜索

此内容没有您所选择的语言版本。

Chapter 1. Overview of Kamelets

download PDF

Kamelets are high-level connectors that can serve as building blocks in an event driven architecture solution. They are custom resources that you can install on an OpenShift cluster and use in Camel K integrations. Kamelets accelerate your development efforts. They simplify how you connect data sources (that emit events) and data sinks (that consume events). Because you configure kamelet parameters rather than writing code, you do not need to be familiar with the Camel DSL to use kamelets.

You can use kamelets to connect applications and services directly to each other or to:

1.1. About kamelets

Kamelets are route components (encapsulated code) that work as connectors in a Camel integration. You can think of kamelets as templates that define where to consume data from (a source) and where to send data to (a sink) - allowing you to assemble data pipelines. Kamelets can also filter, mask, and perform simple calculation logic on data.

There are three different types of kamelets:

  • source - A route that produces data. You use a source kamelet to retrieve data from a component.
  • sink - A route that consumes data. You use a sink kamelet to send data to a component.
  • action - A route that performs an action on data. You can use an action kamelet to manipulate data when it passes from a source kamelet to a sink kamelet.

1.1.1. Why use kamelets?

In a microservices and event-driven architecture solution, kamelets can serve as building blocks for sources that emit events and sinks which consume events.

Kamelets provide abstraction (they hide the complexity of connecting to external systems) and reusability (they are a simple way to reuse code and apply it to different use cases).

Here are some example use cases:

  • You want your application to consume events from Telegram, you can use kamelets to bind the Telegram source to a channel of events. Later, you can connect your application to that channel so that it reacts to those events.
  • You want your application to connect Salesforce directly to Slack.

Kamelets allow you, and your integration development team, to be more efficient. You can reuse kamelets and share them with your team members who can configure instances for their specific needs. The underlying Camel K operator does the hard work: it compiles, builds, packages and deploys the integration defined by the kamelet.

1.1.2. Who uses kamelets?

Because kamelets allow you to reduce the amount of coding you need to do in your Camel integration, they are ideal for developers who are not familiar with the Camel DSL. Kamelets can help smooth the learning curve for a non-Camel developer. There is no need for you to learn another framework or language to get Camel running.

Kamelets are also useful for experienced Camel developers who want to encapsulate complex Camel integration logic into a reusable kamelet, and then share it with other users.

1.1.3. What are the prerequisites for using kamelets?

To use Kamelets, you need the following environment setup:

  • You can access an OpenShift 4.6 (or later) cluster with the correct access level, the ability to create projects and install operators, and the ability to install the OpenShift and Camel K CLI tools on your local system.
  • You installed the Camel K operator in your namespace or cluster-wide as described in Installing Camel K
  • You installed the OpenShift command line (oc) interface tool.
  • Optionally, you installed VS code or another development tool with the Camel K plugin. The Camel-based tooling extensions include features such as automatic completion of Camel URIs based on the embedded kamelet catalog. For more information, see the Camel K development tooling section in Getting Started with Camel K.

    Note: Visual Studio (VS) Code Tooling extensions are community only.

1.1.4. How do you use kamelets?

Using a kamelet typically involves two components: the kamelet itself, which defines a reusable route snippet, and a kamelet binding, in which you reference and bind together one or more kamelets. A kamelet binding is an OpenShift resource (KameletBinding).

Within the kamelet binding resource, you can:

  • Connect a sink or a source kamelet to a channel of events: a Kafka topic or a Knative destination (channel or broker).
  • Connect a sink kamelet directly to a Camel Uniform Resource Identifier (URI). You can also connect a source kamelet to a Camel URI, although connecting a URI and a sink kamelet is the most common use case.
  • Connect a sink and a source kamelet directly to each other, without using a channel of events as a middle-layer.
  • Reference the same kamelet multiple times in the same kamelet binding.
  • Add action kamelets to manipulate data when it passes from a source kamelet to a sink kamelet.
  • Define an error handling strategy to specify what Camel K should do if there is a failure when sending or receiving event data.

At runtime, the Camel K operator uses the kamelet binding to generate and run a Camel K integration.

Note: While Camel DSL developers can use kamelets directly in Camel K integrations, the simpler way to implement kamelets is by specifying a kamelet binding resource to build a high-level event flow.

1.2. Connecting sources and sinks

Use kamelets when you want to connect two or more components (external applications or services). Each kamelet is basically a route template with configuration properties. You need to know which component you want to get data from (a source) and which component you want to send data to (a sink). You connect the source and sink components by adding kamelets in a kamelet binding as illustrated in Figure 1.1.

Kamelet binding source to sink

Figure 1.1: Kamelet binding source to sink

Here is an overview of the steps for using kamelets in a kamelet binding:

  1. Install the Camel K operator. It includes a catalog of kamelets as resources in your OpenShift project.
  2. Create a kamelet binding. Determine which services or applications you want to connect within the kamelet binding.
  3. View the Kamelet Catalog to find the kamelets for the source and sink components that you want to use.
  4. For each kamelet that you want to include in the kamelet binding, determine the configuration properties that you need to set.
  5. In the kamelet binding code, add a reference to each kamelet and configure the required properties.
  6. Apply the kamelet binding as a resource in your OpenShift project.

The Camel K operator uses the kamelet binding to generate and run an integration.

1.2.1. Installing Camel K

You can install the Red Hat Integration - Camel K Operator on your OpenShift cluster from the OperatorHub. The OperatorHub is available from the OpenShift Container Platform web console and provides an interface for cluster administrators to discover and install Operators.

After you install the Camel K Operator, you can install the Camel K CLI tool for command line access to all Camel K features.

Prerequisites

  • You have access to an OpenShift 4.6 (or later) cluster with the correct access level, the ability to create projects and install operators, and the ability to install CLI tools on your local system.

    Note

    You do not need to create a pull secret when installing Camel K from the OpenShift OperatorHub. The Camel K Operator automatically reuses the OpenShift cluster-level authentication to pull the Camel K image from registry.redhat.io.

  • You installed the OpenShift CLI tool (oc) so that you can interact with the OpenShift cluster at the command line. For details on how to install the OpenShift CLI, see Installing the OpenShift CLI.

Procedure

  1. In the OpenShift Container Platform web console, log in by using an account with cluster administrator privileges.
  2. Create a new OpenShift project:

    1. In the left navigation menu, click Home > Project > Create Project.
    2. Enter a project name, for example, my-camel-k-project, and then click Create.
  3. In the left navigation menu, click Operators > OperatorHub.
  4. In the Filter by keyword text box, type Camel K and then click the Red Hat Integration - Camel K Operator card.
  5. Read the information about the operator and then click Install. The Operator installation page opens.
  6. Select the following subscription settings:

    • Update Channel > 1.4.x
    • Installation Mode > A specific namespace on the cluster > my-camel-k-project
    • Approval Strategy > Automatic

      Note

      The Installation mode > All namespaces on the cluster and Approval Strategy > Manual settings are also available if required by your environment.

  7. Click Install, and then wait a few moments until the Camel K Operator is ready for use.
  8. Download and install the Camel K CLI tool:

    1. From the Help menu (?) at the top of the OpenShift web console, select Command line tools.
    2. Scroll down to the kamel - Red Hat Integration - Camel K - Command Line Interface section.
    3. Click the link to download the binary for your local operating system (Linux, Mac, Windows).
    4. Unzip and install the CLI in your system path.
    5. To verify that you can access the Kamel K CLI, open a command window and then type the following:

      kamel --help

      This command shows information about Camel K CLI commands.

1.2.2. Viewing the Kamelet Catalog

When you install the Camel K operator, it includes a catalog of kamelets that you can use in your Camel K integrations.

Prerequisite

You installed the Camel K operator in your working namespace or cluster-wide as described in Installing Camel K.

Procedure

To view a list of kamelets installed with the Camel K operator:

  1. In a Terminal window, login to your OpenShift cluster.
  2. Viewing the list of available kamelets depends on how the Camel K operator was installed (in a specific namespace or cluster-mode):

    • If the Camel K operator is installed in cluster-mode, use this command to view the available kamelets:

      oc get kamelet -n openshift-operators

    • If the Camel K operator is installed in a specific namespace:

      1. Open a project in which the Camel K operator is installed.

        oc project <camelk-project>

        For example, if the Camel K operator is installed in the my-camel-k-project project:

        oc project my-camel-k-project

      2. Run the following command:

        oc get kamelets

Note

For a list of the kamelets that are supported by Red Hat, see the Red Hat Integration Release Notes .

1.2.2.1. Adding a custom kamelet to your Kamelet Catalog

If you don’t see a kamelet in the catalog that suits your requirements, a Camel DSL developer can create a custom kamelet as described in the Apache Camel Kamelets Developers Guide (community documentation). A kamelet is coded in YAML format and, by convention, has a .kamelet.yaml file extension.

Prerequisites

  • A Camel DSL developer has provided you with a custom kamelet file.
  • The kamelet name must be unique to the OpenShift namespace in which the Camel K operator is installed.

Procedure

To make a custom kamelet available as a resource in your OpenShift namespace:

  1. Download the kamelet YAML file (for example, custom-sink.kamelet.yaml) to a local folder.
  2. Login to your OpenShift cluster.
  3. In a Terminal window, open the project in which the Camel K operator is installed, for example my-camel-k-project:

    oc project my-camel-k-project

  4. Run the oc apply command to add the custom kamelet as a resource to the namespace:

    oc apply -f <custom-kamelet-filename>

    For example, use the following command to add the custom-sink.kamelet.yaml file that is located in the current directory:

    oc apply -f custom-sink.kamelet.yaml

  5. To verify that the kamelet is available as a resource, use the following command to view an alphabetical list of all kamelets in the current namespace and then look for your custom kamelet:

    oc get kamelets

1.2.2.2. Determining a kamelet’s configuration parameters

In a kamelet binding, when you add a reference to a kamelet, you specify the name of the kamelet and you configure the kamelet’s parameters.

Prerequisite

  • You installed the Camel K operator in your working namespace or cluster-wide.

Procedure

To determine a kamelet’s name and parameters:

  1. In a terminal window, login to your OpenShift cluster.
  2. Open the kamelet’s YAML file::

    oc describe kamelets/<kamelet-name>

    For example, to view the ftp-source kamelet’s code, if the Camel K operator is installed in the current namespace, use this command:

    oc describe kamelets/ftp-source

    If the Camel K operator is installed in cluster-mode, use this command:

    oc describe -n openshift-operators kamelets/ftp-source

  3. In the YAML file, scroll down to the spec.definition section (which is written in JSON-schema format) to see the list of the kamelet’s properties. At the end of the section, the required field lists the properties that you must configure when you reference the kamelet.

    For example, the following code is an excerpt from the spec.definition section of the ftp-source kamelet. This section provides details for all of the kamelet’s configuration properties. The required properties for this kamelet are connectionHost, connectionPort, username, password, and directoryName:

    spec:
      definition:
        title: "FTP Source"
        description: |-
          Receive data from an FTP Server.
        required:
          - connectionHost
          - connectionPort
          - username
          - password
          - directoryName
        type: object
        properties:
          connectionHost:
            title: Connection Host
            description: Hostname of the FTP server
            type: string
          connectionPort:
            title: Connection Port
            description: Port of the FTP server
            type: string
            default: 21
          username:
            title: Username
            description: The username to access the FTP server
            type: string
          password:
            title: Password
            description: The password to access the FTP server
            type: string
            format: password
            x-descriptors:
            - urn:alm:descriptor:com.tectonic.ui:password
          directoryName:
            title: Directory Name
            description: The starting directory
            type: string
          passiveMode:
            title: Passive Mode
            description: Sets passive mode connection
            type: boolean
            default: false
            x-descriptors:
            - 'urn:alm:descriptor:com.tectonic.ui:checkbox'
          recursive:
            title: Recursive
            description: If a directory, will look for files in all the sub-directories as well.
            type: boolean
            default: false
            x-descriptors:
            - 'urn:alm:descriptor:com.tectonic.ui:checkbox'
          idempotent:
            title: Idempotency
            description: Skip already processed files.
            type: boolean
            default: true
            x-descriptors:
            - 'urn:alm:descriptor:com.tectonic.ui:checkbox'

1.2.3. Connecting source and sink components in a kamelet binding

Within a kamelet binding, you connect source and sink components.

The example in this procedure uses the following kamelets as shown in Figure 1.2:

  • The example source kamelet is named coffee-source. This simple kamelet retrieves randomly-generated data about types of coffee from a web site catalog. It has one parameter (period - an integer value) that determines how frequently (in seconds) to retrieve the coffee data. The parameter is not required since there is a default value (1000 seconds).
  • The example sink kamelet is named log-sink. It retrieves data and outputs it to a log file.
Example kamelet binding

Figure 1.2: Example kamelet binding

Prerequisites

  • You know how to create and edit a Camel K integration.
  • The Red Hat Integration - Camel K operator is installed on your OpenShift namespace or cluster and you have downloaded the Red Hat Integration Camel K CLI tool as described in Installing Camel K.
  • You know which kamelets you want to add to your Camel K integration and their required instance parameters.
  • The kamelets that you want to use are available in your Kamelet catalog.

    If you want to use the kamelets in this example, copy and save the coffee-source code to a local file named coffee-source.kamelet.yaml and the log-sink code to a file named log-sink.kamelet.yaml. Then run the following commands to add them to your Kamelet catalog:

    oc apply -f coffee-source.kamelet.yaml

    oc apply -f log-sink.kamelet.yaml

Procedure

  1. Login to your OpenShift cluster.
  2. Open your working project where the Camel K operator is installed. If you installed the Camel K operator in cluster-mode, it is available to any project on the cluster.

    For example, to open an existing project named my-camel-k-project:

    oc project my-camel-k-project

  3. Create a new KameletBinding resource:

    1. In an editor of your choice, create a YAML file with the following structure:

      apiVersion: camel.apache.org/v1alpha1
      kind: KameletBinding
      metadata:
        name:
      spec:
        source:
        sink:
    2. Add a name for the kamelet binding.

      For this example, the name is coffee-to-log because the binding connects the coffee-source kamelet to the log-sink kamelet.

      apiVersion: camel.apache.org/v1alpha1
      kind: KameletBinding
      metadata:
        name: coffee-to-log
      spec:
        source:
        sink:
  4. Specify the source kamelet (for example, coffee-source) and configure any parameters for the kamelet.

    Note: For this example, the parameter is defined within the kamelet binding’s YAML file. Alternatively, you can configure a kamelet’s parameters in a property file, ConfigMap, or Secret as described in Configuring kamelet instance parameters.

    apiVersion: camel.apache.org/v1alpha1
    kind: KameletBinding
    metadata:
      name: coffee-to-log
    spec:
      source:
        ref
          kind: Kamelet
          apiVersion: camel.apache.org/v1alpha1
          name: coffee-source
        properties:
          period: 5000
      sink:
  5. Specify the sink kamelet (for example, log-sink) and configure any parameters for the kamelet. The example log-sink kamelet does not have any required parameters.

    apiVersion: camel.apache.org/v1alpha1
    kind: KameletBinding
    metadata:
      name: coffee-to-log
    spec:
      source:
        ref:
          kind: Kamelet
          apiVersion: camel.apache.org/v1alpha1
          name: coffee-source
        properties:
          period: 5000
      sink:
        ref:
          kind: Kamelet
          apiVersion: camel.apache.org/v1alpha1
          name: log-sink
  6. Save the YAML file (for example, coffee-to-log.yaml).
  7. Add the KameletBinding as a resource to your OpenShift namespace:

    oc apply -f <kamelet>.yaml

    For example:

    oc apply -f coffee-to-log.yaml

    The Camel K operator generates and runs a Camel K integration by using the KameletBinding resource.

  8. To see the status of the KameletBinding:

    oc get kameletbindings

  9. To see the status of the corresponding integration: oc get integrations
  10. To view the output:

    • To view the logs from the command line, open a Terminal window and then type the following command:

      kamel log <integration-name>

      For example, if the integration name is coffee-to-log, use this command:

      kamel log coffee-to-log

    • To view the logs from OpenShift web console:

      1. Select Workloads > Pods.
      2. Click the name of the Camel K integration’s pod, and then click Logs.

        You should see a list of coffee events similar to the following example:

        INFO  [log-sink-E80C5C904418150-0000000000000001] (Camel (camel-1) thread #0 - timer://tick) {"id":7259,"uid":"a4ecb7c2-05b8-4a49-b0d2-d1e8db5bc5e2","blend_name":"Postmodern Symphony","origin":"Huila, Colombia","variety":"Kona","notes":"delicate, chewy, black currant, red apple, star fruit","intensifier":"balanced"}
  11. To stop the integration, delete the kamelet binding:

    oc delete kameletbindings/<kameletbinding-name>

    For example:

    oc delete kameletbindings/coffee-to-log

Next steps

Optionally:

1.2.4. Configuring kamelet instance parameters

When you reference a kamelet, you have the following options for defining the kamelet’s instance parameters:

  • Directly in a kamelet binding where you specify the kamelet URI. In the following example, the bot authorization token provided by the Telegram BotFather. is 123456:

    from("kamelet:telegram-source?authorizationToken=123456")

  • Globally configure a kamelet property (so that you don’t have to provide the value in the URI) by using the following format:

    "camel.kamelet.<kamelet-name>.<property-name>=<value>”

    As described in the Configuring Camel K integrations chapter in Developing and Managing Integrations Using Camel K, you can configure kamelet parameters by:

    • Defining them as properties
    • Defining them in a property file
    • Defining them in an OpenShift ConfigMap or Secret

1.2.5. Connecting to a channel of events

The most common use case for kamelets is to use a kamelet binding to connect them to a channel of events: a Kafka topic or a Knative destination (channel or broker). The advantage of doing so is that the data source and sink are independent and “unaware” of each other. This decoupling allows the components in your business scenario to be developed and managed separately. If you have multiple data sinks and sources as part of your business scenario, it becomes even more important to decouple the various components. For example, if an event sink needs to be shut down, the event source is not impacted. And, if other sinks use the same source, they are not impacted.

Figure 1.3 illustrates the flow of connecting source and sink kamelets to a channel of events.

Connecting source and sink kamelets to a channel of events

Figure 1.3: Connecting source and sink kamelets to a channel of events

If you use the Apache Kafka stream-processing framework, for details on how to connect to a Kafka topic, see Connecting to Kafka with Kamelets.

If you use the Knative serverless framework, for details on how to connect to a Knative destination (channel or broker), see Connecting to Knative with Kamelets.

1.2.6. Connecting to an explicit Camel URI

You can create a kamelet binding in which a kamelet sends events to—or receives events from—an explicit Camel URI. Typically, you bind a source kamelet to a URI that can receive events (that is, you specify the URI as the sink in a kamelet binding). Examples of Camel URIs that receive events are HTTP or HTTPS endpoints.

It is also possible, but not as common, to specify a URI as the source in a kamelet binding. Examples of Camel URIs that send events are timer, mail, or FTP endpoints.

To connect a kamelet to a Camel URI, follow the steps in Connecting source and sink components in a kamelet binding and for the sink.uri field, instead of a kamelet, specify an explicit Camel URI.

In the following example, the URI for the sink is a fictional URI (https://mycompany.com/event-service):

apiVersion: camel.apache.org/v1alpha1
kind: KameletBinding
metadata:
  name: coffee-to-event-service
spec:
  source:
    ref:
      kind: Kamelet
      apiVersion: camel.apache.org/v1alpha1
      name: coffee-source
    properties:
      period: 5000
  sink:
    uri: https://mycompany.com/event-service

1.3. Applying operations to data within a connection

If you want to perform an operation on the data that passes between a kamelet and an event channel, use action kamelets as intermediary steps within a kamelet binding. For example, you can use an action kamelet to serialize or deserialize data, filter the data, or insert a field or a message header.

Manipulation operations, such as filtering or adding fields, work only with JSON data (that is, when the Content-Type header is set to application/json). If the event data uses a format other than JSON (for example, Avro or Protocol Buffers), you must convert the format of the data by adding a deserialize step (for example, that references the protobuf-deserialize-action or avro-deserialize-action kamelet) before the manipulating action and a serialize step (for example, that references the protobuf-serialize-action or avro-serialize-action kamelet) after it. For more information about converting the format of data in a connection, see Data conversion kamelets.

Action kamelets include:

1.3.1. Adding an operation to a kamelet binding

To implement an action kamelet, in the kamelet binding file’s spec section, add a steps section in between the source and sink sections.

Prerequisites

  • You have created a kamelet binding as described in Connecting source and sink components in a kamelet binding.
  • You know which action kamelet you want to add to the kamelet binding and the action kamelet’s required parameters.

    For the example in this procedure, the parameter for the predicate-filter-action kamelet is a string type, expression, that provides the JSON Path Expression that filters coffee data to only log coffees that have a “deep” taste intensity. Note that the predicate-filter-action kamelet requires that you set a Builder trait configuration property in the kamelet binding.

    The example also includes deserialize and serialize actions which are optional in this case because the event data format is JSON.

Procedure

  1. Open a KameletBinding file in an editor.

    For example, here are the contents of the coffee-to-log.yaml file:

    apiVersion: camel.apache.org/v1alpha1
    kind: KameletBinding
    metadata:
      name: coffee-to-log
    spec:
      source:
        ref:
          kind: Kamelet
          apiVersion: camel.apache.org/v1alpha1
          name: coffee-source
        properties:
          period: 5000
      sink:
        ref:
          kind: Kamelet
          apiVersion: camel.apache.org/v1alpha1
          name: log-sink
  2. Add an integration section above the source section and provide the following Builder trait configuration property (as required by the predicate-filter-action kamelet):

    apiVersion: camel.apache.org/v1alpha1
    kind: KameletBinding
    metadata:
      name: coffee-to-log
    spec:
      integration:
             traits:
               builder:
                 configuration:
                   properties:
                     - "quarkus.arc.unremovable- types=com.fasterxml.jackson.databind.ObjectMapper"
      source:
        ref:
          kind: Kamelet
          apiVersion: camel.apache.org/v1alpha1
          name: coffee-source
        properties:
          period: 5000
      sink:
        ref:
          kind: Kamelet
          apiVersion: camel.apache.org/v1alpha1
          name: log-sink
  3. Add a steps section, between the source and sink sections and define the action kamelet. For example:

    apiVersion: camel.apache.org/v1alpha1
    kind: KameletBinding
    metadata:
      name: coffee-to-log
      spec:
        integration:
               traits:
                 builder:
                   configuration:
                     properties:
                       - "quarkus.arc.unremovable-types=com.fasterxml.jackson.databind.ObjectMapper"
      source:
        ref:
          kind: Kamelet
          apiVersion: camel.apache.org/v1alpha1
          name: coffee-source
        properties:
          period: 5000
      steps:
      - ref:
          kind: Kamelet
          apiVersion: camel.apache.org/v1alpha1
          name: json-deserialize-action
      - ref:
          kind: Kamelet
          apiVersion: camel.apache.org/v1alpha1
          name: predicate-filter-action
        properties:
          expression: "@.intensifier =~ /.*deep/"
      - ref:
          kind: Kamelet
          apiVersion: camel.apache.org/v1alpha1
          name: json-serialize-action
      sink:
        ref:
          kind: Kamelet
          apiVersion: camel.apache.org/v1alpha1
          name: log-sink
  4. Save your changes.
  5. Update the KameletBinding resource:

    oc apply -f coffee-to-log.yaml

    The Camel K operator re-generates and runs the CamelK integration that it generates based upon the updated KameletBinding resource.

  6. To see the status of the kamelet binding:

    oc get kameletbindings

  7. To see the status of its corresponding integration:

    oc get integrations

  8. To view the log file output:

    1. Get the pod name for the integration:

      oc get pods

    2. View the log for the integration’s pod:

      oc logs <podname of the integration>

      For example, if the pod name is example-7885bdb9-84ht8:

      oc logs example-7885bdb9-84ht8

  9. To stop the integration, delete the kamelet binding:

    oc delete kameletbindings/<kameletbinding-name>

    For example:

    oc delete kameletbindings/coffee-to-log

1.3.2. Action kamelets

1.3.2.1. Data filtering kamelets

You can filter the data that passes between source and sink components, for example, to prevent leaking sensitive data or to avoid generating unnecessary networking charges.

You can filter data based on the following criteria:

  • Kafka topic name - Filter events for a Kafka topic with a name that matches the given Java regular expression by configuring the Topic Name Matches Filter Action Kamelet (topic-name-matches-filter-action).
  • Header key - Filter events that have a given message header by configuring the Header Filter Action Kamelet (has-header-filter-action).
  • Null value - Filters tombstone events (events with a null payload) by configuring the Tombstone Filter Action Kamelet (is-tombstone-filter-action).
  • Predicate - Filter events based on the given JSON path expression by configuring the Predicate Filter Action Kamelet (predicate-filter-action). The predicate-filter-action kamelet requires that you set the following Builder trait configuration property in the kamelet binding:

    spec:
      integration:
        traits:
          builder:
            configuration:
              properties:
               - "quarkus.arc.unremovable-types=com.fasterxml.
                    jackson.databind.ObjectMapper"
Note

Data filtering kamelets work out-of-the-box with JSON data (that is, when the Content-Type header is set to application/json). If the event data uses a format other than JSON, you must convert the format of the data by adding a deserialize step (for example, protobuf-deserialize-action or avro-deserialize-action) before the manipulating action and a serialize step (for example, protobuf-serialize-action or avro-serialize-action) after it. For more information about converting the format of data in a connection, see Data conversion kamelets.

1.3.2.2. Data conversion kamelets

With the following data conversion kamelets, you can serialize and deserialize the format of data that passes between source and sink components. The data conversion applies to the payload of event data (not the key or the header).

  • Avro - An open source project that provides data serialization and data exchange services for Apache Hadoop.

    • Avro Deserialize Action Kamelet (avro-deserialize-action)
    • Avro Serialize Action Kamelet (avro-serialize-action)
  • Protocol Buffers - A high-performance, compact binary wire format invented by Google who use it internally so they can communicate with their internal network services.

    • Protobuf Deserialize Action Kamelet (protobuf-deserialize-action)
    • Protobuf Serialize Action Kamelet (protobuf-serialize-action)
  • JSON (JavaScript Object Notation) - A data-interchange format that is based on a subset of the JavaScript Programming Language. JSON is a text format that is completely language independent.

    • JSON Deserialize Action Kamelet (json-deserialize-action)
    • JSON Serialize Action Kamelet (json-serialize-action)
Note

You must specify the schema (as a single-line, using JSON format) in the Avro and Protobuf serialize/deserialize kamelets. You do not need to do so for JSON serialize/deserialize kamelets.

1.3.2.3. Data transformation kamelets

With the following data transformation kamelets, you can perform simple manipulations on the data that passes between the source and sink components:

  • ValueToKey - (for Kafka) Use the value-to-key-action kamelet to replace the record key with a new key formed from a subset of fields in the payload. You can set the event key to a value that is based on the event information before the data is written to Kafka. For example, when reading records from a database table, you can partition the records in Kafka based on the customer ID.
  • MaskField - Use the mask-field-action kamelet to replace a field value with a valid null value for the field type (such as 0 or an empty string) or with a given replacement (the replacement must be a non-empty string or a numeric value).

    For example, if you want to capture data from a relational database to send to Kafka and the data includes protected (PCI / PII) information, you must mask the protected information if your Kafka cluster is not certified yet.

  • InsertHeader - Use the insert-header-action kamelet to add a field (header) by using either static data or record metadata.
  • InsertField - Use the insert-field-action kamelet to add a field (value) by using either static data or record metadata.

1.4. Handling errors within a connection

To specify what the Camel K operator should do if a running integration encounters a failure when sending or receiving event data, you can optionally add one of the following error handling policies to the kamelet binding:

  • No error handler - Ignores any failure happening in your integration.
  • Log error handler - Sends a log message to standard output.
  • Dead letter channel error handler - Redirects a failing event to another component, such as a third-party URI, a queue, or another kamelet which can perform certain logic with the failing event. Also supports attempting to redeliver the message exchange a number of times before sending it to a dead letter endpoint.
  • Bean error handler - Specifies to use a custom bean for handling errors.
  • Ref error handler - Specifies to use a bean for handling errors. The bean must be available in the Camel registry at runtime.

1.4.1. Adding an error handler policy to a kamelet binding

To handle errors when sending or receiving event data between a source and a sink connection, add an error handler policy to the kamelet binding.

Prerequisites

  • You know which type of error handler policy you want to use.
  • You have an existing KameletBinding YAML file.

Procedure

To implement error handling in a kamelet binding:

  1. Open a KameletBinding YAML file in an editor.
  2. Add an error handler section to the spec section, after the sink definition:

    apiVersion: camel.apache.org/v1alpha1
    kind: KameletBinding
    metadata:
      name: example-kamelet-binding
    spec:
      source:
       ...
      sink:
       ...
      errorHandler: ...

    For example, in the coffee-to-log kamelet binding, specify the maximum number of times an error is sent to the log file by adding a log error handler:

    apiVersion: camel.apache.org/v1alpha1
    kind: KameletBinding
    metadata:
      name: coffee-to-log
    spec:
      source:
        ref:
          kind: Kamelet
          apiVersion: camel.apache.org/v1alpha1
          name: coffee-source
        properties:
          period: 5000
      sink:
        ref:
          kind: Kamelet
          apiVersion: camel.apache.org/v1alpha1
          name: log-sink
      errorHandler: log: parameters: maximumRedeliveries: 3
  3. Save your file.

1.4.2. Error handlers

1.4.2.1. No error handler

If you want to ignore any failure happening in your integration, you can either not include an errorHandler section in the kamelet binding or set it to none as shown in the following example:

apiVersion: camel.apache.org/v1alpha1
kind: KameletBinding
metadata:
  name: my-kamelet-binding
spec:
  source:
...
  sink:
...
  errorHandler:
    none:

1.4.2.2. Log error handler

The default behavior for handling any failure is to send a log message to standard output. Optionally, you can use the log error handler to specify other behaviors, such as a redelivery or delay policy, as shown in the following example:

apiVersion: camel.apache.org/v1alpha1
kind: KameletBinding
metadata:
  name: my-kamelet-binding
spec:
  source:
...
  sink:
...
  errorHandler:
    log: parameters: maximumRedeliveries: 3 redeliveryDelay: 2000

1.4.2.3. Dead letter channel error handler

The Dead Letter Channel allows you to redirect any failing event to any other component (such as a third party URI, a queue, or another kamelet) that can define how to handle a failing event, as shown in the following example:

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: 2
          message: "ERROR!"
          ...
      parameters: 3
        maximumRedeliveries: 1
  1. For the endpoint, you can use ref or uri. The Camel K operator interprets ref according to the kind, apiVersion and name values. You can use any kamelet, Kafka Topic channel, or Knative destination.
  2. Properties that belong to the endpoint (in this example, to a kamelet named error-handler).
  3. Parameters that belong to the dead-letter-channel error handler type.

1.4.2.4. Bean error handler

With the Bean error handler you can extend the functionality of the Error Handler by providing a custom bean that handles errors. For type, specify the fully-qualified name of the ErrorHandlerBuilder. For properties, configure the properties expected by the ErrorHandlerBuilder that you specified in type.

Note

Within a kamelet binding, Camel components in URIs require dependency declarations. The Camel K operator generates an integration from a kamelet binding at runtime. For integrations that are not generated by a kamelet binding, Camel K automatically handles the dependency management and imports all the required libraries from the Camel catalog. However, for this release, you must specify any Camel components that you reference in a URI within a kamelet binding as a dependency. In the following example, because the kamelet binding references the camel-log component in the deadLetterUri, it includes the camel-log as a dependency in the spec.integration.dependencies section.

apiVersion: camel.apache.org/v1alpha1
kind: KameletBinding
metadata:
  name: my-kamelet-binding
spec:
  integration: dependencies: - camel:log
  source:

  source:
...
  sink:
...
  errorHandler:
    bean: type: "org.apache.camel.builder.DeadLetterChannelBuilder" properties: deadLetterUri: log:error

1.4.2.5. Ref error handler

With the Ref error handler, you can use any bean that you expect to be available in the Camel registry at runtime. In the following example, my-custom-builder is the name of the bean to look up at runtime.

apiVersion: camel.apache.org/v1alpha1
kind: KameletBinding
metadata:
  name: my-kamelet-binding
spec:
  source:
   ...
  sink:
   ...
  errorHandler:
    ref: my-custom-builder
Red Hat logoGithubRedditYoutubeTwitter

学习

尝试、购买和销售

社区

关于红帽文档

通过我们的产品和服务,以及可以信赖的内容,帮助红帽用户创新并实现他们的目标。

让开源更具包容性

红帽致力于替换我们的代码、文档和 Web 属性中存在问题的语言。欲了解更多详情,请参阅红帽博客.

關於紅帽

我们提供强化的解决方案,使企业能够更轻松地跨平台和环境(从核心数据中心到网络边缘)工作。

© 2024 Red Hat, Inc.