9.6. Configuring Kafka Connect

Procedure

1. Edit the authorization property in the KafkaUser resource to provide access rights to the user.

   Access rights are configured for the Kafka Connect topics and cluster group using literal name values. The following table shows the default names configured for the topics and cluster group ID.

   Expand

   表 9.2. Names for the access rights configuration

   Property	Name
   offset.storage.topic	connect-cluster-offsets
   status.storage.topic	connect-cluster-status
   config.storage.topic	connect-cluster-configs
   group	connect-cluster

   Show more

   In this example configuration, the default names are used to specify access rights. If you are using different names for a Kafka Connect instance, use those names in the ACLs configuration.

   Example configuration for simple authorization

   apiVersion: kafka.strimzi.io/v1beta2
   kind: KafkaUser
   metadata:
     name: my-user
     labels:
       strimzi.io/cluster: my-cluster
   spec:
     # ...
     authorization:
       type: simple
       acls:
         # access to offset.storage.topic
         - resource:
             type: topic
             name: connect-cluster-offsets
             patternType: literal
           operations:
             - Create
             - Describe
             - Read
             - Write
           host: "*"
         # access to status.storage.topic
         - resource:
             type: topic
             name: connect-cluster-status
             patternType: literal
           operations:
             - Create
             - Describe
             - Read
             - Write
           host: "*"
         # access to config.storage.topic
         - resource:
             type: topic
             name: connect-cluster-configs
             patternType: literal
           operations:
             - Create
             - Describe
             - Read
             - Write
           host: "*"
         # cluster group
         - resource:
             type: group
             name: connect-cluster
             patternType: literal
           operations:
             - Read
           host: "*"

2. Create or update the resource.

   oc apply -f KAFKA-USER-CONFIG-FILE

Procedure

1. Find the name of the KafkaConnector custom resource that controls the connector you want to pause or stop:

   oc get KafkaConnector

2. Edit the KafkaConnector resource to stop or pause the connector.

   Example configuration for stopping a Kafka Connect connector

   apiVersion: kafka.strimzi.io/v1beta2
   kind: KafkaConnector
   metadata:
     name: my-source-connector
     labels:
       strimzi.io/cluster: my-connect-cluster
   spec:
     class: org.apache.kafka.connect.file.FileStreamSourceConnector
     tasksMax: 2
     config:
       file: "/opt/kafka/LICENSE"
       topic: my-topic
     state: stopped
     # ...

   Change the state configuration to stopped or paused. The default state for the connector when this property is not set is running.

3. Apply the changes to the KafkaConnector configuration.

   You can resume the connector by changing state to running or removing the configuration.

Procedure

1. Find the name of the KafkaConnector custom resource that controls the Kafka connector you want to restart:

   oc get KafkaConnector

2. Restart the connector by annotating the KafkaConnector resource in OpenShift:

   oc annotate KafkaConnector <kafka_connector_name> strimzi.io/restart="true"

   The restart annotation is set to true.

3. Wait for the next reconciliation to occur (every two minutes by default).

   The Kafka connector is restarted, as long as the annotation was detected by the reconciliation process. When Kafka Connect accepts the restart request, the annotation is removed from the KafkaConnector custom resource.

Procedure

1. Find the name of the KafkaConnector custom resource that controls the Kafka connector task you want to restart:

   oc get KafkaConnector

2. Find the ID of the task to be restarted from the KafkaConnector custom resource:

   oc describe KafkaConnector <kafka_connector_name>

   Task IDs are non-negative integers, starting from 0.

3. Use the ID to restart the connector task by annotating the KafkaConnector resource in OpenShift:

   oc annotate KafkaConnector <kafka_connector_name> strimzi.io/restart-task="0"

   In this example, task 0 is restarted.

4. Wait for the next reconciliation to occur (every two minutes by default).

   The Kafka connector task is restarted, as long as the annotation was detected by the reconciliation process. When Kafka Connect accepts the restart request, the annotation is removed from the KafkaConnector custom resource.