Configure Kafka Connect

Kafka Connect (Connect) streams data between Apache Kafka® and other data systems using connectors, which define where data is copied to and from.

Note

Centralized Connect License does not work with the Docker image confluentinc/cp-kafka-connect. Use the confluentinc/cp-server-connect Docker image to leverage the Centralized License.

When you deploy a Connect cluster using Confluent for Kubernetes (CFK), it gets deployed in the distributed mode. CFK does not support the standalone mode deployment of the Connect cluster.

On a Confluent Private Cloud multi-tenant Physical Kafka Cluster (PKC), Connect resolves topic names that are local to a single Logical Kafka Cluster (LKC). Deploy a dedicated Connect cluster for each LKC rather than sharing one across LKCs. To point a Connect CR at an LKC instead of the PKC, set the following in spec.dependencies.kafka.discovery:

kind: LogicalKafkaCluster. Set it explicitly. If you omit it, CFK looks for a Kafka CR and a LogicalKafkaCluster CR that have the same name, and fails if both exist.
name: The name of the LogicalKafkaCluster CR.
namespace: The namespace of the LogicalKafkaCluster CR.
secretRef: The name of the Secret that holds the credentials that Connect uses to authenticate to the PKC’s in-cluster token listener. CFK can’t resolve the LKC without it.

For more information, see Deploy Ecosystem Components on a Multi-Tenant PKC.

Note

Connect deployment through CFK does not support GCP Application Load Balancer (ALB) or static host-based routing for external access. For the supported external access methods, see Configure Networking. For an example of host-based ingress for Confluent Platform components, see the ingress-service-hostbased-example.yaml in the CFK examples repository.

Connector plugins are the binaries or JARs that implement the classes and abstractions of connectors. Connector plugins are installed in Connect workers.

When configuring Connect, you need to specify how to install connector plugins for the Connect deployment.

You have the following options to install connector plugins in CFK:

Automatically download and install connector plugins.
Extend the Connect Docker image with connector plugins.

After setting the connector info and other required settings, deploy Connect with kubectl apply.

Starting in CFK 2.1.0, you can declaratively manage connectors in Kubernetes using the Connector custom resource definition (CRD). After you install the connector plugins as described in this topic, see Work with Connectors for managing connectors.

Install connector plugin

Automatically download and install connector plugins

CFK can automatically download and install connector plugins or JARs from Confluent Hub or a custom artifacts location URL. By specifying both options in the same Connect custom resource (CR), you can download multiple connectors from Confluent Hub and from custom URLs.

To store the connector plugins, you need to have a specified size of node volume available. The default size is 4 GB, but you can specify a different size in the Connect CR using storageLimit.

Note

If you are setting the custom plugin.path property in spec.configOverrides (for example, to use the FileStream connectors), you must include /mnt/plugins in plugin.path. This applies whether you download plugins from Confluent Hub or from a custom URL. For example:

spec:
  configOverrides:
    server:
    - plugin.path=/usr/share/java,/mnt/plugins

Download connector plugins from Confluent Hub

Provide the download info in the Connect CR as follows:

kind: Connect
metadata:
  annotations:
    platform.confluent.io/confluent-hub-install-extra-args:    --- [1]
spec:
  build:
    type: onDemand                                             --- [2]
    onDemand:
      plugins:
        confluentHub:                                          --- [3]
          - name:                                              --- [4]
            owner:                                             --- [5]
            version:                                           --- [6]
      storageLimit:                                            --- [7]

[1] Optional. An annotation for the additional arguments to be used when the Connect starts up and downloads plugins from Confluent Hub. For example:
```
platform.confluent.io/confluent-hub-install-extra-args: "--worker-configs /dev/null --component-dir /mnt/plugins"
```
[2] Required to have CFK automatically download connector plugins.
[3] Provide an array of plugins to be downloaded.
[4] Required. The name of this connector plugin.
[5] Required. The individual or organization that provides the plugin, for example, confluentinc.
[6] Required. The version of this plugin. Set to the version of the plugin or latest.
[7] Optional. The maximum amount of node volume that can be used to store the connector plugins. The default value is 4 GB.

Download connector plugins from a custom URL

Provide the download information in the Connect CR as follows. Only the relevant properties related to the Connect custom URL are described here. For a complete CR example, see CFK GitHub examples repo.

Note

When downloading plugins from a custom URL, CFK first attempts the download with wget. If wget fails for some reason, CFK falls back to curl.

connector-wget-extra-args: Passes extra arguments to wget.
connector-curl-extra-args: Passes extra arguments to the curl fallback.
connector-force-curl: Skips the wget attempt entirely and downloads plugins with curl directly.

kind: Connect
metadata:
  annotations:
    platform.confluent.io/connector-wget-extra-args:   --- [1]
    platform.confluent.io/connector-curl-extra-args:   --- [2]
    platform.confluent.io/connector-force-curl:        --- [3]
spec:
  build:
    type: onDemand                                   --- [4]
    onDemand:                                        --- [5]
      plugins:
        url:                                         --- [6]
        - name:                                      --- [7]
          archivePath:                               --- [8]
          checksum:                                  --- [9]
      storageLimit:                                  --- [10]

[1] Optional. An annotation for the additional arguments to be used when the Connect starts up and downloads plugins from a custom URL with wget.

Set the annotation to "--no-check-certificate" to ignore SSL verification when downloading plugins from a site with SSL enabled but when the certificate authority (CA) is not available to CFK:

Caution

This annotation is not recommended as it could pose a security risk.
[2] Optional. An annotation for the additional arguments to be used when CFK falls back to curl after a wget failure while downloading plugins from a custom URL. Use this annotation to pass any curl options, such as proxy settings, authentication headers, or SSL flags.

For example:
- Set the annotation to "-k" or "--insecure" to ignore SSL verification when downloading plugins from a site with SSL enabled but when the CA is not available to CFK.
  
  Caution
  
  This option is not recommended as it could pose a security risk.
- Set the annotation to "-x http://proxy.example.com:8080" to route the download through an HTTP proxy.
- Set the annotation to '-H "Authorization: Bearer <token>"' to pass an authentication header when downloading plugins from a site that requires authentication.
[3] Optional. Set to "true" to skip the wget attempt entirely and have Connect download plugins from the custom URL with curl directly.
[4] Required to have CFK automatically download connector plugins.
[5] Required when type: onDemand is set in [4].

[6] An array of plugins to be downloaded. Each plugin must specify the values for name: ([7]), archivePath: ([8]), and checksum: ([9]).

For example:

plugins:
  url:
  - name: kafka-connect-jdbc
    archivePath: https://raw.githubusercontent.com/confluentinc/confluent-kubernetes-examples/master/hybrid/ccloud-JDBC-mysql/confluentinc-kafka-connect-jdbc-10.2.5.zip
    checksum: 9d033fabac89ec0b35a97246f7ca3a36800bcb402ccfbd76adebd4c4c9ca6e7d6a044a9162383a18f99400893581b55a87b9fe89e35276507e9a3fb6cff3fda0

[7] Required. The name of this connector plugin.
[8] Required. The archive path of the zip file that contains this plugin.
[9] Required. Defines the sha512sum checksum of the plugin’s remote file. It is used to verify the remote file after download.
[10] Optional. The maximum amount of node volume that can be used to store the connector plugins. The default value is 4G.

View Connect init container logs

To troubleshoot issues while installing connector plugins, run the kubectl logs command to view the logs from the init container of the Connect pod. For example:

kubectl logs -f connect-0 -c config-init-container

Extend the Connect Docker image with connector plugins

To extend the Connect image with connector plugins, build a new Docker image that includes the connectors, then reference that image in the Connect CR.

Add a new connector to one of the following Connect images:

For Confluent Platform 6.2.x and later, use the cp-server-connect image.

The remainder of this document uses this image.
For Confluent Platform 6.1.x and earlier, use the cp-server-connect-operator image.

The image contains Connect and all of its dependencies. It does not contain any connector JARs.

Follow these steps to build and reference the new image:

Create a Dockerfile in <dockerfile-dir> to add one or more connectors to the cp-server-connect image.

You can either:

Pull connectors from Confluent Hub.
Use the connector JAR downloaded to the machine you are running the Docker build from.

To pull connectors from Confluent Hub:

Create a Dockerfile as follows:

FROM confluentinc/cp-server-connect:<Confluent Platform release>
USER root
RUN confluent-hub install --no-prompt <connector1>:<connector1-version> \
  && confluent-hub install --no-prompt <connector2>:<connector2-version> \
  && ...
USER 1001

An example Dockerfile to create a Docker image with the data-gen connector from Confluent Hub:

FROM confluentinc/cp-server-connect:8.3.0
USER root
RUN  confluent-hub install --no-prompt confluentinc/kafka-connect-datagen:0.3.3
USER 1001

To use the connector JAR downloaded to the machine you are running the Docker build from:

Create a Dockerfile as follows:

FROM confluentinc/cp-server-connect:<Confluent Platform release>
ADD <local-connector1-path> /usr/share/java/<connector1> \
  && <local-connector2-path> /usr/share/java/<connector2> \
  && ...
USER 1001

An example Dockerfile to use the data-gen connector existing on your local machine in the <connector-dir> directory:

FROM confluentinc/cp-server-connect:8.3.0
ADD my-connector-dir/confluentinc-kafka-connect-datagen /usr/share/java/confluentinc-kafka-connect-datagen
USER 1001

Build and push the image with the following commands:

docker build <dockerfile-dir> -t <someregistry>/<somerepository>:<sometag>

docker push <someregistry>/<somerepository>:<sometag>

Get the Docker image details from the output of the preceding process and specify the repository and tag in the Connect CR.
```
spec:
  image:
    application: <someregistry>/<somerepository>:<sometag>
```

Provide connector certificates

To make certificates available to connectors in the Connect cluster, use a Kubernetes secret or HashiCorp Vault to provide the certificates, as described in the following steps.

If using a Kubernetes secret for the certificates, create the secret using the kubectl create secret command, and pass the secret name in the next step in secretRef ([4]).
If using Vault for the certificates, inject the certificates as described in Provide secrets in HashiCorp Vault, and pass the directory path in the next step in directoryPathInContainer ([2]).

For example use cases, see CFK examples for using Vault.
Specify the secret name in the Connect CR:
```
kind: Connect
spec:
  connectorTLSCerts:          --- [1]
  - directoryPathInContainer: --- [2]
    jksPassword:
      secretRef:              --- [3]
    secretRef:                --- [4]
```
- [1] A list of connector TLS certificate references injected in the Connect pod for the connector use.
  
  Use one of directoryPathInContainer, jksPassword, or secretRef for each set of certificate information.
- [2] [4] CFK supports the secretRef and directoryPathInContainer methods to inject the connector TLS certificates. Specify either directoryPathInContainer ([2]) or secretRef ([4]).
- [2] The directory path in container where keystore.jks, truststore.jks, jksPassword.txt keys are mounted. For example:
```
spec:
  connectorTLSCerts:
  - directoryPathInContainer: /vault/secrets
```
  When you use the Vault directory path (directoryPathInContainer), CFK does not automate the creation of keystore and truststore. You need to create the keystore and truststore first, and those must be present in the Vault directory path.
- [3] The secret name referenced for the JKS password. Expects the key-value in the following format: jksPassword.txt=jksPassword=<user_provided_password>
  
  If omitted, CFK uses the default password, mystorepassword. For more information, see Provide TLS keys and certificates in Java KeyStore format.
- [4] The Kubernetes secret name that contains the connector TLS certificates.
In the connector CR, specify the locations.

See Connector TLS certificates for an example.

For an example configuration, see Replicator Connector Configuration.

Provide Connect credentials using mounted secrets

You can use mounted secrets to protect sensitive data, such as passwords, in connector configuration.

Create a secret as described in Mount custom Kubernetes secrets. For example:
```
kubectl create secret generic my-credential \
  --from-file=my-credential.txt=/my-dir/my-credential.txt
```
The secret reference is mounted in the default path, /mnt/secrets/<secret-name>.

Specify the preceding secret name in the Connect CR:

spec:
  mountedSecrets:
    - secretRef:  # The name of the secret that contains the credentials.

In the connector CR, you specify the locations of the secret as variables, and CFK dynamically resolves the variables when the connector starts.

See Mounted secrets for credentials for an example.