主页
产品
Streams for Apache Kafka
2.7
Deploying and Managing Streams for Apache Kafka on OpenShift
6.5. Adding Kafka Connect connectors

6.5. Adding Kafka Connect connectors

Kafka Connect uses connectors to integrate with other systems to stream data. A connector is an instance of a Kafka Connector class, which can be one of the following type:

Source connector: A source connector is a runtime entity that fetches data from an external system and feeds it to Kafka as messages.
Sink connector: A sink connector is a runtime entity that fetches messages from Kafka topics and feeds them to an external system.

Kafka Connect uses a plugin architecture to provide the implementation artifacts for connectors. Plugins allow connections to other systems and provide additional configuration to manipulate data. Plugins include connectors and other components, such as data converters and transforms. A connector operates with a specific type of external system. Each connector defines a schema for its configuration. You supply the configuration to Kafka Connect to create a connector instance within Kafka Connect. Connector instances then define a set of tasks for moving data between systems.

Add connector plugins to Kafka Connect in one of the following ways:

Configure Kafka Connect to build a new container image with plugins automatically
Create a Docker image from the base Kafka Connect image (manually or using continuous integration)

After plugins have been added to the container image, you can start, stop, and manage connector instances in the following ways:

You can also create new connector instances using these options.

6.5.1. Building a new container image with connector plugins automatically
复制链接

Configure Kafka Connect so that Streams for Apache Kafka automatically builds a new container image with additional connectors. You define the connector plugins using the .spec.build.plugins property of the KafkaConnect custom resource. Streams for Apache Kafka will automatically download and add the connector plugins into a new container image. The container is pushed into the container repository specified in .spec.build.output and automatically used in the Kafka Connect deployment.

Prerequisites

The Cluster Operator must be deployed.
A container registry.

You need to provide your own container registry where images can be pushed to, stored, and pulled from. Streams for Apache Kafka supports private container registries as well as public registries such as Quay or Docker Hub.

Procedure

Configure the KafkaConnect custom resource by specifying the container registry in .spec.build.output, and additional connectors in .spec.build.plugins:

apiVersion: kafka.strimzi.io/v1beta2
kind: KafkaConnect
metadata:
  name: my-connect-cluster
spec:


  #...
  build:
    output:


      type: docker
      image: my-registry.io/my-org/my-connect-cluster:latest
      pushSecret: my-registry-credentials
    plugins:


      - name: connector-1
        artifacts:
          - type: tgz
            url: <url_to_download_connector_1_artifact>
            sha512sum: <SHA-512_checksum_of_connector_1_artifact>
      - name: connector-2
        artifacts:
          - type: jar
            url: <url_to_download_connector_2_artifact>
            sha512sum: <SHA-512_checksum_of_connector_2_artifact>
  #...

1: The specification for the Kafka Connect cluster.
2: (Required) Configuration of the container registry where new images are pushed.
3: (Required) List of connector plugins and their artifacts to add to the new container image. Each plugin must be configured with at least one artifact.

Create or update the resource:

$ oc apply -f <kafka_connect_configuration_file>

Wait for the new container image to build, and for the Kafka Connect cluster to be deployed.
Use the Kafka Connect REST API or KafkaConnector custom resources to use the connector plugins you added.

6.5.2. Building a new container image with connector plugins from the Kafka Connect base image
复制链接

Create a custom Docker image with connector plugins from the Kafka Connect base image. Add the custom image to the /opt/kafka/plugins directory.

You can use the Kafka container image on Red Hat Ecosystem Catalog as a base image for creating your own custom image with additional connector plugins.

At startup, the Streams for Apache Kafka version of Kafka Connect loads any third-party connector plugins contained in the /opt/kafka/plugins directory.

Prerequisites

The Cluster Operator must be deployed.

Procedure

Create a new Dockerfile using registry.redhat.io/amq-streams/kafka-37-rhel9:2.7.0 as the base image:

FROM registry.redhat.io/amq-streams/kafka-37-rhel9:2.7.0
USER root:root
COPY ./my-plugins/ /opt/kafka/plugins/
USER 1001

Example plugins file

$ tree ./my-plugins/
./my-plugins/
├── debezium-connector-mongodb
│   ├── bson-<version>.jar
│   ├── CHANGELOG.md
│   ├── CONTRIBUTE.md
│   ├── COPYRIGHT.txt
│   ├── debezium-connector-mongodb-<version>.jar
│   ├── debezium-core-<version>.jar
│   ├── LICENSE.txt
│   ├── mongodb-driver-core-<version>.jar
│   ├── README.md
│   └── # ...
├── debezium-connector-mysql
│   ├── CHANGELOG.md
│   ├── CONTRIBUTE.md
│   ├── COPYRIGHT.txt
│   ├── debezium-connector-mysql-<version>.jar
│   ├── debezium-core-<version>.jar
│   ├── LICENSE.txt
│   ├── mysql-binlog-connector-java-<version>.jar
│   ├── mysql-connector-java-<version>.jar
│   ├── README.md
│   └── # ...
└── debezium-connector-postgres
    ├── CHANGELOG.md
    ├── CONTRIBUTE.md
    ├── COPYRIGHT.txt
    ├── debezium-connector-postgres-<version>.jar
    ├── debezium-core-<version>.jar
    ├── LICENSE.txt
    ├── postgresql-<version>.jar
    ├── protobuf-java-<version>.jar
    ├── README.md
    └── # ...

The COPY command points to the plugin files to copy to the container image.

This example adds plugins for Debezium connectors (MongoDB, MySQL, and PostgreSQL), though not all files are listed for brevity. Debezium running in Kafka Connect looks the same as any other Kafka Connect task.

Build the container image.
Push your custom image to your container registry.
Point to the new container image.
You can point to the image in one of the following ways:
- Edit the KafkaConnect.spec.image property of the KafkaConnect custom resource.
  If set, this property overrides the STRIMZI_KAFKA_CONNECT_IMAGES environment variable in the Cluster Operator.
  apiVersion: kafka.strimzi.io/v1beta2 kind: KafkaConnect metadata: name: my-connect-cluster spec:
  1
  #... image: my-new-container-image
  2
  config:
  3
  #...
  1
  The specification for the Kafka Connect cluster.
  2
  The docker image for Kafka Connect pods.
  3
  Configuration of the Kafka Connect workers (not connectors).
- Edit the STRIMZI_KAFKA_CONNECT_IMAGES environment variable in the install/cluster-operator/060-Deployment-strimzi-cluster-operator.yaml file to point to the new container image, and then reinstall the Cluster Operator.

6.5.3. Deploying KafkaConnector resources
复制链接

Deploy KafkaConnector resources to manage connectors. The KafkaConnector custom resource offers an OpenShift-native approach to management of connectors by the Cluster Operator. You don’t need to send HTTP requests to manage connectors, as with the Kafka Connect REST API. You manage a running connector instance by updating its corresponding KafkaConnector resource, and then applying the updates. The Cluster Operator updates the configurations of the running connector instances. You remove a connector by deleting its corresponding KafkaConnector.

KafkaConnector resources must be deployed to the same namespace as the Kafka Connect cluster they link to.

In the configuration shown in this procedure, the autoRestart feature is enabled (enabled: true) for automatic restarts of failed connectors and tasks. You can also annotate the KafkaConnector resource to restart a connector or restart a connector task manually.

Example connectors

You can use your own connectors or try the examples provided by Streams for Apache Kafka. Up until Apache Kafka 3.1.0, example file connector plugins were included with Apache Kafka. Starting from the 3.1.1 and 3.2.0 releases of Apache Kafka, the examples need to be added to the plugin path as any other connector.

Streams for Apache Kafka provides an example KafkaConnector configuration file (examples/connect/source-connector.yaml) for the example file connector plugins, which creates the following connector instances as KafkaConnector resources:

A FileStreamSourceConnector instance that reads each line from the Kafka license file (the source) and writes the data as messages to a single Kafka topic.
A FileStreamSinkConnector instance that reads messages from the Kafka topic and writes the messages to a temporary file (the sink).

We use the example file to create connectors in this procedure.

注意

The example connectors are not intended for use in a production environment.

Prerequisites

A Kafka Connect deployment
The Cluster Operator is running

Procedure

Add the FileStreamSourceConnector and FileStreamSinkConnector plugins to Kafka Connect in one of the following ways:
- Configure Kafka Connect to build a new container image with plugins automatically
- Create a Docker image from the base Kafka Connect image (manually or using continuous integration)

Set the strimzi.io/use-connector-resources annotation to true in the Kafka Connect configuration.

apiVersion: kafka.strimzi.io/v1beta2
kind: KafkaConnect
metadata:
  name: my-connect-cluster
  annotations:
    strimzi.io/use-connector-resources: "true"
spec:
    # ...

With the KafkaConnector resources enabled, the Cluster Operator watches for them.

Edit the examples/connect/source-connector.yaml file:
Example KafkaConnector source connector configuration
```
apiVersion: kafka.strimzi.io/v1beta2
kind: KafkaConnector
metadata:
  name: my-source-connector  
```
1
```
  labels:
    strimzi.io/cluster: my-connect-cluster 
```
2
```
spec:
  class: org.apache.kafka.connect.file.FileStreamSourceConnector 
```
3
```
  tasksMax: 2 
```
4
```
  autoRestart: 
```
5
```
    enabled: true
  config: 
```
6
```
    file: "/opt/kafka/LICENSE" 
```
7
```
    topic: my-topic 
```
8
```
    # ...
```
1
Name of the KafkaConnector resource, which is used as the name of the connector. Use any name that is valid for an OpenShift resource.
2
Name of the Kafka Connect cluster to create the connector instance in. Connectors must be deployed to the same namespace as the Kafka Connect cluster they link to.
3
Full name of the connector class. This should be present in the image being used by the Kafka Connect cluster.
4
Maximum number of Kafka Connect tasks that the connector can create.
5
Enables automatic restarts of failed connectors and tasks. By default, the number of restarts is indefinite, but you can set a maximum on the number of automatic restarts using the maxRestarts property.
6
Connector configuration as key-value pairs.
7
Location of the external data file. In this example, we’re configuring the FileStreamSourceConnector to read from the /opt/kafka/LICENSE file.
8
Kafka topic to publish the source data to.
Create the source KafkaConnector in your OpenShift cluster:
```
oc apply -f examples/connect/source-connector.yaml
```
Create an examples/connect/sink-connector.yaml file:
```
touch examples/connect/sink-connector.yaml
```

Paste the following YAML into the sink-connector.yaml file:

apiVersion: kafka.strimzi.io/v1beta2
kind: KafkaConnector
metadata:
  name: my-sink-connector
  labels:
    strimzi.io/cluster: my-connect
spec:
  class: org.apache.kafka.connect.file.FileStreamSinkConnector


  tasksMax: 2
  config:


    file: "/tmp/my-file"


    topics: my-topic

1: Full name or alias of the connector class. This should be present in the image being used by the Kafka Connect cluster.
2: Connector configuration as key-value pairs.
3: Temporary file to publish the source data to.
4: Kafka topic to read the source data from.

Create the sink KafkaConnector in your OpenShift cluster:
```
oc apply -f examples/connect/sink-connector.yaml
```
Check that the connector resources were created:
```
oc get kctr --selector strimzi.io/cluster=<my_connect_cluster> -o name

my-source-connector
my-sink-connector
```
Replace <my_connect_cluster> with the name of your Kafka Connect cluster.
In the container, execute kafka-console-consumer.sh to read the messages that were written to the topic by the source connector:
```
oc exec <my_kafka_cluster>-kafka-0 -i -t -- bin/kafka-console-consumer.sh --bootstrap-server <my_kafka_cluster>-kafka-bootstrap.NAMESPACE.svc:9092 --topic my-topic --from-beginning
```
Replace <my_kafka_cluster> with the name of your Kafka cluster.

Source and sink connector configuration options

The connector configuration is defined in the spec.config property of the KafkaConnector resource.

The FileStreamSourceConnector and FileStreamSinkConnector classes support the same configuration options as the Kafka Connect REST API. Other connectors support different configuration options.

Expand

表 6.1. Configuration options for the FileStreamSource connector class
Name	Type	Default value	Description
`file`	String	Null	Source file to write messages to. If not specified, the standard input is used.
`topic`	List	Null	The Kafka topic to publish data to.

Expand

表 6.2. Configuration options for FileStreamSinkConnector class
Name	Type	Default value	Description
`file`	String	Null	Destination file to write messages to. If not specified, the standard output is used.
`topics`	List	Null	One or more Kafka topics to read data from.
`topics.regex`	String	Null	A regular expression matching one or more Kafka topics to read data from.

6.5.4. Exposing the Kafka Connect API
复制链接

Use the Kafka Connect REST API as an alternative to using KafkaConnector resources to manage connectors. The Kafka Connect REST API is available as a service running on <connect_cluster_name>-connect-api:8083, where <connect_cluster_name> is the name of your Kafka Connect cluster. The service is created when you create a Kafka Connect instance.

The operations supported by the Kafka Connect REST API are described in the Apache Kafka Connect API documentation.

注意

The strimzi.io/use-connector-resources annotation enables KafkaConnectors. If you applied the annotation to your KafkaConnect resource configuration, you need to remove it to use the Kafka Connect API. Otherwise, manual changes made directly using the Kafka Connect REST API are reverted by the Cluster Operator.

You can add the connector configuration as a JSON object.

Example curl request to add connector configuration

curl -X POST \
  http://my-connect-cluster-connect-api:8083/connectors \
  -H 'Content-Type: application/json' \
  -d '{ "name": "my-source-connector",
    "config":
    {
      "connector.class":"org.apache.kafka.connect.file.FileStreamSourceConnector",
      "file": "/opt/kafka/LICENSE",
      "topic":"my-topic",
      "tasksMax": "4",
      "type": "source"
    }
}'

The API is only accessible within the OpenShift cluster. If you want to make the Kafka Connect API accessible to applications running outside of the OpenShift cluster, you can expose it manually by creating one of the following features:

LoadBalancer or NodePort type services
Ingress resources (Kubernetes only)
OpenShift routes (OpenShift only)

注意

The connection is insecure, so allow external access advisedly.

If you decide to create services, use the labels from the selector of the <connect_cluster_name>-connect-api service to configure the pods to which the service will route the traffic:

Selector configuration for the service

# ...
selector:
  strimzi.io/cluster: my-connect-cluster


  strimzi.io/kind: KafkaConnect
  strimzi.io/name: my-connect-cluster-connect


#...

1: Name of the Kafka Connect custom resource in your OpenShift cluster.
2: Name of the Kafka Connect deployment created by the Cluster Operator.

You must also create a NetworkPolicy that allows HTTP requests from external clients.

Example NetworkPolicy to allow requests to the Kafka Connect API

apiVersion: networking.k8s.io/v1
kind: NetworkPolicy
metadata:
  name: my-custom-connect-network-policy
spec:
  ingress:
  - from:
    - podSelector:


        matchLabels:
          app: my-connector-manager
    ports:
    - port: 8083
      protocol: TCP
  podSelector:
    matchLabels:
      strimzi.io/cluster: my-connect-cluster
      strimzi.io/kind: KafkaConnect
      strimzi.io/name: my-connect-cluster-connect
  policyTypes:
  - Ingress

1: The label of the pod that is allowed to connect to the API.

To add the connector configuration outside the cluster, use the URL of the resource that exposes the API in the curl command.

6.5.5. Limiting access to the Kafka Connect API
复制链接

It is crucial to restrict access to the Kafka Connect API only to trusted users to prevent unauthorized actions and potential security issues. The Kafka Connect API provides extensive capabilities for altering connector configurations, which makes it all the more important to take security precautions. Someone with access to the Kafka Connect API could potentially obtain sensitive information that an administrator may assume is secure.

The Kafka Connect REST API can be accessed by anyone who has authenticated access to the OpenShift cluster and knows the endpoint URL, which includes the hostname/IP address and port number.

For example, suppose an organization uses a Kafka Connect cluster and connectors to stream sensitive data from a customer database to a central database. The administrator uses a configuration provider plugin to store sensitive information related to connecting to the customer database and the central database, such as database connection details and authentication credentials. The configuration provider protects this sensitive information from being exposed to unauthorized users. However, someone who has access to the Kafka Connect API can still obtain access to the customer database without the consent of the administrator. They can do this by setting up a fake database and configuring a connector to connect to it. They then modify the connector configuration to point to the customer database, but instead of sending the data to the central database, they send it to the fake database. By configuring the connector to connect to the fake database, the login details and credentials for connecting to the customer database are intercepted, even though they are stored securely in the configuration provider.

If you are using the KafkaConnector custom resources, then by default the OpenShift RBAC rules permit only OpenShift cluster administrators to make changes to connectors. You can also designate non-cluster administrators to manage Streams for Apache Kafka resources. With KafkaConnector resources enabled in your Kafka Connect configuration, changes made directly using the Kafka Connect REST API are reverted by the Cluster Operator. If you are not using the KafkaConnector resource, the default RBAC rules do not limit access to the Kafka Connect API. If you want to limit direct access to the Kafka Connect REST API using OpenShift RBAC, you need to enable and use the KafkaConnector resources.

For improved security, we recommend configuring the following properties for the Kafka Connect API:

org.apache.kafka.disallowed.login.modules

(Kafka 3.4 or later) Set the org.apache.kafka.disallowed.login.modules Java system property to prevent the use of insecure login modules. For example, specifying com.sun.security.auth.module.JndiLoginModule prevents the use of the Kafka JndiLoginModule.

Example configuration for disallowing login modules

apiVersion: kafka.strimzi.io/v1beta2
kind: KafkaConnect
metadata:
  name: my-connect-cluster
  annotations:
    strimzi.io/use-connector-resources: "true"
spec:
  # ...
  jvmOptions:
    javaSystemProperties:
      - name: org.apache.kafka.disallowed.login.modules
        value: com.sun.security.auth.module.JndiLoginModule, org.apache.kafka.common.security.kerberos.KerberosLoginModule
# ...

Only allow trusted login modules and follow the latest advice from Kafka for the version you are using. As a best practice, you should explicitly disallow insecure login modules in your Kafka Connect configuration by using the org.apache.kafka.disallowed.login.modules system property.

connector.client.config.override.policy

Set the connector.client.config.override.policy property to None to prevent connector configurations from overriding the Kafka Connect configuration and the consumers and producers it uses.

Example configuration to specify connector override policy

apiVersion: kafka.strimzi.io/v1beta2
kind: KafkaConnect
metadata:
  name: my-connect-cluster
  annotations:
    strimzi.io/use-connector-resources: "true"
spec:
  # ...
  config:
    connector.client.config.override.policy: None
# ...

6.5.6. Switching from using the Kafka Connect API to using KafkaConnector custom resources
复制链接

You can switch from using the Kafka Connect API to using KafkaConnector custom resources to manage your connectors. To make the switch, do the following in the order shown:

Deploy KafkaConnector resources with the configuration to create your connector instances.
Enable KafkaConnector resources in your Kafka Connect configuration by setting the strimzi.io/use-connector-resources annotation to true.

警告

If you enable KafkaConnector resources before creating them, you delete all connectors.

To switch from using KafkaConnector resources to using the Kafka Connect API, first remove the annotation that enables the KafkaConnector resources from your Kafka Connect configuration. Otherwise, manual changes made directly using the Kafka Connect REST API are reverted by the Cluster Operator.

When making the switch, check the status of the KafkaConnect resource. The value of metadata.generation (the current version of the deployment) must match status.observedGeneration (the latest reconciliation of the resource). When the Kafka Connect cluster is Ready, you can delete the KafkaConnector resources.

6.5. Adding Kafka Connect connectors

6.5.1. Building a new container image with connector plugins automatically
复制链接

6.5.2. Building a new container image with connector plugins from the Kafka Connect base image
复制链接

6.5.3. Deploying KafkaConnector resources
复制链接

Source and sink connector configuration options

6.5.4. Exposing the Kafka Connect API
复制链接

6.5.5. Limiting access to the Kafka Connect API
复制链接

6.5.6. Switching from using the Kafka Connect API to using KafkaConnector custom resources
复制链接

学习

尝试、购买和销售

社区

關於紅帽

让开源更具包容性

关于红帽文档

Theme

Red Hat legal and privacy links

Red Hat legal and privacy links

6.5. Adding Kafka Connect connectors

6.5.1. Building a new container image with connector plugins automatically复制链接链接已复制到粘贴板!

6.5.2. Building a new container image with connector plugins from the Kafka Connect base image复制链接链接已复制到粘贴板!

6.5.3. Deploying KafkaConnector resources复制链接链接已复制到粘贴板!

Source and sink connector configuration options

6.5.4. Exposing the Kafka Connect API复制链接链接已复制到粘贴板!

6.5.5. Limiting access to the Kafka Connect API复制链接链接已复制到粘贴板!

6.5.6. Switching from using the Kafka Connect API to using KafkaConnector custom resources复制链接链接已复制到粘贴板!

学习

尝试、购买和销售

社区

關於紅帽

让开源更具包容性

关于红帽文档

Theme

Red Hat legal and privacy links

Red Hat legal and privacy links

6.5.1. Building a new container image with connector plugins automatically
复制链接

6.5.2. Building a new container image with connector plugins from the Kafka Connect base image
复制链接

6.5.3. Deploying KafkaConnector resources
复制链接

6.5.4. Exposing the Kafka Connect API
复制链接

6.5.5. Limiting access to the Kafka Connect API
复制链接

6.5.6. Switching from using the Kafka Connect API to using KafkaConnector custom resources
复制链接