After completing this section, you should be able to create a Kafka Connect cluster.

Kafka Connect is a tool for reliably streaming data between Apache Kafka and external systems. It is both scalable and flexible. You can use Kafka Connect to integrate Kafka with any system, such as databases, key-value stores, search indexes and file systems.

Figure 5.1: High-level Kafka Connect workflow

Without Kafka Connect, developers must build their own applications to integrate Kafka with an external system. For each integration point, there must be a custom application that performs the data transfer between the external systems. The custom applications do not share a custom common contract unless it is implemented. Implementing a custom contract that provides reusability and flexibility is not easy. It also creates the risk of reliability and scalability issues because of the possible defects in the custom integration applications.

Kafka Connect consists of two main components: the Kafka Connect cluster and the Kafka Connect connectors.

A Kafka Connect Cluster consists of one or multiple nodes called workers. This is very similar to a Kafka cluster because the workload can be distributed across the nodes.

Figure 5.2: Kafka Connect workers

Connectors are covered in more detail in the following section.

A Kafka Connect cluster, without connectors, does not do anything by itself. The process that runs in the Kafka Connect node, will just run idly. The cluster's main responsibility is to manage the worker nodes and ensure that the connector workload per task is distributed evenly. To transfer data between Kafka and an external system, there must be actively running connector tasks.

Kafka Connect Working Modes

Kafka Connect has two working modes:

Standalone Mode

In standalone mode, only a single Kafka Connect worker is responsible for execution. This execution runs all the connectors and their tasks in the same worker. Thus, this mode is not suitable for building a highly available, scalable system. It is more suitable as a local adapter that solves a minor integration problem.

With this mode running, Kafka offsets and Kafka Connect configuration is stored locally in files.

Figure 5.3: Kafka Connect running in standalone mode

Distributed Mode

In distributed mode, there are multiple workers to distribute the workload of connector tasks. Like a Kafka cluster, a Kafka Connect cluster has a leader that is one of the worker nodes. The Leader is responsible for coordinating and balancing the task distribution for all worker nodes including itself.

Distributed mode provides scalability and automatic fault tolerance for Kafka Connect. This helps to build a highly available system which is important for most of the production environments.

Because the nodes are distributed, you must configure a Kafka Connect cluster for keeping the configuration and the offsets stored in Kafka topics. In this way, all the workers can sync the configuration and offset data.

Figure 5.4: Kafka Connect running in distributed mode

In AMQ Streams on the Red Hat OpenShift Container Platform, Kafka Connect Clusters always run in distributed mode. The worker nodes run in containers instead of machines and can scale with the help of pods. Each pod has one Kafka Connect container that has one worker in it.

Because Kafka Connect runs in distributed mode, connector offsets, configurations and status are all stored in Kafka topics.

In AMQ Streams on the Red Hat OpenShift Container Platform (RHOCP), Kafka Connect Clusters are configured by using the KafkaConnect custom resource. The following YAML definition is a basic example of a Kafka Connect cluster configuration with two replicas.

apiVersion: kafka.strimzi.io/v1beta2
kind: KafkaConnect 
metadata:
  annotations:
    strimzi.io/use-connector-resources: 'true' 
  name: my-connect-cluster 
spec:
  bootstrapServers: 'my-cluster-kafka-bootstrap:9092' 
  image: 'quay.io/redhattraining/connect-cluster:latest' 
  config: 
    config.storage.topic: my-connect-cluster-configs 
    offset.storage.topic: my-connect-cluster-offsets 
    status.storage.topic: my-connect-cluster-status 
    config.storage.replication.factor: 1
    offset.storage.replication.factor: 1
    status.storage.replication.factor: 1
  replicas: 2 
  version: 2.8.0 

The KafkaConnect.spec.image value from the preceding example is a custom, prebuilt Kafka Connect cluster image. This is a manually built image that the Kafka Connect cluster runs in its containers. In most cases, you should create a custom image to include the required plug-ins for the connectors. Plug-ins are the .jar files that connectors use when they are active in a Kafka Connect cluster.

Manually building an image is not the only way to create a Kafka Connect cluster. To prepare a Kafka Connect cluster image, there are two options:

In this lecture, we will focus on building a cluster image with the AMQ Streams operator.

apiVersion: kafka.strimzi.io/v1beta2
kind: KafkaConnect
metadata:
  annotations:
    strimzi.io/use-connector-resources: 'true'
  name: my-connect-cluster
spec:
  bootstrapServers: my-cluster-kafka-bootstrap:9092
  build: 
    output: 
      image: quay.io/redhattraining/connect-cluster:latest 
      pushSecret: my-connect-cluster-push-secret 
      type: docker 
    plugins: 
    - name: debezium-postgres-connector
      artifacts:
      - type: tgz 
        url: https://a.plugin.url/debezium-connector-postgres-1.3.1.Final-plugin.tar.gz 
        sha512sum: 962a12151bdf9a5a30627...f035a0447cb820077af00c03 
    - name: camel-elasticsearch-rest-kafka-connector
      artifacts:
      - type: tgz
        url: https://a.plugin.url/camel-elasticsearch-rest-kafka-connector-0.10.0-package.tar.gz
  config:
    config.storage.topic: my-connect-cluster-configs
    offset.storage.topic: my-connect-cluster-offsets
    status.storage.topic: my-connect-cluster-status
    config.storage.replication.factor: 1
    offset.storage.replication.factor: 1
    status.storage.replication.factor: 1
  replicas: 2
  version: 2.8.0

Apart from providing a set of custom resources that the AMQ Streams operator provides, Kafka Connect also offers a REST API. This REST API allows developers to get the worker node info and manage connectors on the Kafka Connect cluster. In AMQ Streams on RHOCP, this is available on connect-cluster-name-connect-api:8083. This address is exposed via a Service that the operator creates.

The top-level resource paths for the REST API are:

/ is the root path for the REST API, which gets the information from the worker node and serves the REST request. This information has the version of the Kafka Connect worker, the git commit ID of the source code, and the Kafka Connect cluster ID.

The following is an example of a request and its response to a Kafka Connect cluster's REST API for its root path.

GET / HTTP/1.1
Host: connect-cluster-name-connect-api
Accept: application/json

HTTP/1.1 200 OK
Content-Type: application/json

{
  "version":"2.8.0.redhat-00005",
  "commit":"5568dc65ed44500b"
  "kafka_cluster_id":"pkAfTbvQTuqfo10Hlg_PxQ"
}

The REST API currently only supports application/json as the content type for both the request and the response. Thus, the requests should specify the expected content type of the response via the Accept header. In the response header, the Content-Type should be application/json as well, if it is specified in the request header.

There is more detailed information about connectors and using the REST API for them in the next section named Creating Connectors.