<a id="co-manage-connectors"></a>

# Manage Connectors for Confluent Platform Using Confluent for Kubernetes

Confluent for Kubernetes (CFK) supports declaratively creating and managing connectors as
Connector custom resources (CRs) in Kubernetes.

Connectors are managed by Kafka Connect. Install connector plugins as
described in [Install connector plugin](co-configure-connect.md#co-install-connector-plugin) before you create connector CRs.

<a id="co-create-connectors"></a>

## Create connectors

Each Connector CR is mapped to a connector in Kafka Connect worker.

#### NOTE
When RBAC is enabled in this Confluent Platform environment, the super user you configured
for Kafka (`kafka.spec.authorization.superUsers`) does not have access to
resources in the Connect cluster. If you want the super user to be able to
create connectors, grant the super user the permission on the Connect
cluster.

1. Create a connector as a Connector CR:
   ```yaml
   metadata:
     name:                --- [1]
   kind: Connector
   spec:
     name:                --- [2]
     taskMax:             --- [3]
     class:               --- [4]
     configs:             --- [5]
     restartPolicy:
       type:              --- [6]
       maxRetry:          --- [7]
     connectClusterRef:   --- [8]
     connectRest:         --- [9]
   ```

   * [1] Required. The name of this Connector CR.

     To copy the settings of a connector to create a new connector, change this
     value with the name of the new connector.
   * [2] Optional. Typically, this is the same as `metadata.name`. If you
     need to have a name which Kubernetes cannot support, specify that name
     here.
   * [3] Required. The max number of tasks for the connector.

     When you set this, consider resource consumption. You may need to test in
     a development environment, first, if setting this to a big number.
   * [4] Required. The class name of the connector.
   * [5] Connector-specific configuration settings as key-value maps. Consult
     the connector documentation for the required settings.

     See [Connector configs](#co-connector-configs) for more information, if configuration
     requires sensitive data or certificates.
   * [6] Required. The policy to restart failed tasks of the connector. Set to
     `OnFailure` or `Never`. The default value is `OnFailure`, which means
     it will automatically restart the task  when a task failed if `maxRetry`
     is not reached.
   * [7] The max retry times to restart when `restartPolicy` type is
     `OnFailure`. The default value is `10`.
   * [8] See [Discover Connect using Connect cluster name](#co-connector-discover-using-name).
   * [9] See [Discover Connect using Connect endpoint](#co-connector-discover-using-endpoint).
2. Apply the configuration to create the connector:
   ```bash
   kubectl apply -f <connector CR>
   ```

For  a list of Connector CR examples, see the [sample connector CRs in CFK
Example Rep](https://github.com/confluentinc/confluent-kubernetes-examples/tree/master/connector).

<a id="co-connector-configs"></a>

### Connector configs

You can use connector configs (`spec.configs`) for categories of settings:

* Connector-specific settings
* [Sensitive information, such as user credentials](#co-connector-configs-credentials)
* [Connector TLS certificates](#co-connector-configs-certificates)

<a id="co-connector-configs-credentials"></a>

#### Mounted secrets for credentials

[Mounted secrets](co-credentials.md#co-mounted-secrets) allows you to protect sensitive
data, such as passwords, in connector configuration.

To use mounted secrets for connectors:

1. Create a mounted secret and specify the secret name in the Connect CR as
   described in [Provide Connect credentials using mounted secrets](co-configure-connect.md#co-connect-mounted-secrets).
2. In the connector CR, specify the locations of the secret as variables,
   and CFK dynamically resolves the variables when the connector starts.

   For example, with a secret named, `my-credential`:
   ```yaml
   kind: Connector
   spec:
     configs:
       connection.url: "${file:/mnt/secrets/my-credential/custom.properties:connector-url}"
       connection.user: "${file:/mnt/secrets/my-credential/custom.properties:connector-username}"
       connection.password: "${file:/mnt/secrets/my-credential/custom.properties:connector-password}"
   ```

<a id="co-connector-configs-certificates"></a>

#### Connector TLS certificates

When required connector TLS certificates are not present in the Connect
cluster, follow the steps to have the certificates available to the connectors
in the Connect cluster.

1. Create a secret and specify the secret name in the Connect CR as described
   in [Provide connector certificates](co-configure-connect.md#co-connector-certs).
2. Get the details of the certificates, such as keystore path, truststore path,
   JKS password, using the following command:
   ```bash
   kubectl get connect -oyaml
   ```

   The paths are in `status.connectorTLSFilePaths`.
3. In the Connector CR, specify the information retrieved in the previous
   step.

   For example:
   ```yaml
   spec:
     configs:
       ssl.truststore.location: "/mnt/sslcerts/truststore.jks"
       ssl.truststore.password: "${file:/mnt/sslcerts/jksPassword.txt:jksPassword}"
       ssl.keystore.location: "/mnt/sslcerts/keystore.jks"
       ssl.keystore.password: "${file:/mnt/sslcerts/jksPassword.txt:jksPassword}"
   ```

For an example configuration, see [Replicator Connector Configuration](https://github.com/confluentinc/confluent-kubernetes-examples/tree/master/connector/replicator-source-connector).

### Discover Connect cluster

A connector can discover which Connect cluster it belongs to using one of the
following methods:

* [Provide the name of the Connect cluster](#co-connector-discover-using-name) in `connectClusterRef`.

  Use this option when the Connect cluster is deployed by CFK as a custom
  resource.
* [Provide the connection information for the Connect cluster](#co-connector-discover-using-endpoint) in `connectRest`.

  Use this option when the Connect cluster is deployed outside of CFK.
* If neither `connectClusterRef` or `connectRest` is provided in the
  connector CR, CFK tries to discover the Connect cluster in the same
  namespace. If there are multiple Connect clusters in the same namespace,
  CFK will logs errors.

<a id="co-connector-discover-using-name"></a>

#### Discover Connect using Connect cluster name

To have the connector discover its Connect cluster by the Connect cluster
name, use `connectClusterRef` in the Connector CR:

```yaml
spec:
  connectClusterRef:
    name:                 --- [1]
    namespace:            --- [2]
```

* [1] Required. The name of the Connect cluster.
* [2] Optional. The namespace where the Connect cluster is deployed. If
  omitted, the same namespace as this connector is assumed.

<a id="co-connector-discover-using-endpoint"></a>

#### Discover Connect using Connect endpoint

You can provide information of the Connect cluster that the connector belongs
to in the Connector CR.

**Kafka Connect endpoint**

```yaml
spec:
  connectRest:
    endpoint:             --- [1]
```

* [1] Connect URL and the port.

For example:

```yaml
spec:
  connectRest:
    endpoint: https://connect.operator.svc.cluster.local:8083
```

### Authenticate to Connect cluster

The following authentication methods are supported for a connector to authenticate to a Connect cluster: basic, bearer, mTLS, OAuth. Configure the Connnector CR as described below for the authentication method you want to use.

**Basic authentication to Connect**

```yaml
spec:
  connectRest:
    authentication:
      type: basic                 --- [1]
      basic:
        secretRef:                --- [2]
        directoryPathInContainer: --- [3]
```

* [1] Required.
* [2] or [3] is required.
* [2] The name of the secret that contains the credentials.
  See [Basic authentication](co-authenticate-cp.md#co-authenticate-cp-basic) for the required format.
* [3] The directory path in the container where required authentication
  credentials are injected by Vault.

  See [Basic authentication](co-authenticate-cp.md#co-authenticate-cp-basic) for the required  format.

  See  [Provide secrets for Confluent Platform application CR](co-credentials.md#co-vault-category-2) for providing the credential and required
  annotations when using Vault.

**mTLS authentication to Connect**

```yaml
spec:
  connectRest:
    authentication:
      type: mtls                 --- [1]
    tls:
      secretRef:                 --- [2]
      directoryPathInContainer:  --- [3]
```

* [1] Required.
* [2] or [3] is required.
* [2] The name of the secret that contains the certificates.

  See [Provide TLS keys and certificates in PEM format](co-network-encryption.md#co-certs-pem) for the expected keys in the TLS secret. Only the PEM
  format is supported for Connector CRs.
* [3] The directory path in the container where the required certificates are
  injected by Vault.

  See [Configure Network Encryption for Confluent Platform Using Confluent for Kubernetes](co-network-encryption.md#co-network-encryption) for more information on certificates.

  See  [Provide secrets for Confluent Platform application CR](co-credentials.md#co-vault-category-2) for providing the certificates and required
  annotations when using Vault.

**Bearer authentication to Connect (for RBAC)**

When RBAC is enabled, you can configure bearer authentication as below:

```yaml
spec:
  connectRest:
    authentication:
      type: bearer                --- [1]
      bearer:
        secretRef:                --- [2]
        directoryPathInContainer: --- [3]
```

* [1] Required.
* [2] Required. The name of the secret that contains the bearer credentials. See
  [Bearer authentication](co-authenticate-kafka.md#co-authenticate-mds-bearer) for the required format.
* [3] The directory path in the container where bearer credentials are injected
  by Vault.

  See [Bearer authentication](co-authenticate-kafka.md#co-authenticate-mds-bearer) for the required format.

  See  [Provide secrets for Confluent Platform application CR](co-credentials.md#co-vault-category-2) for providing the credential and required annotations when using Vault.

**OAuth authorization and authentication to Connect**

```yaml
spec:
  connectRest:
    authentication:
      type: oauth                 --- [1]
      oauth:
        secretRef:                --- [2]
        directoryPathInContainer: --- [3]
        configuration:            --- [4]
```

* [1] Required.
* [2] or [3] is required.
* [2] The name of the secret that contains the credentials.
  See [Client-side OAuth/OIDC authentication for Confluent components](co-authenticate-cp.md#co-authenticate-cp-client-oauth) for the required format.
* [3] The directory path in the container where required authentication
  credentials are injected by Vault.

  See [Client-side OAuth/OIDC authentication for Confluent components](co-authenticate-cp.md#co-authenticate-cp-client-oauth) for the required format.

  See  [Provide secrets for Confluent Platform application CR](co-credentials.md#co-vault-category-2) for providing the credential and required
  annotations when using Vault.
* [4] Required. The OAuth settings. For information about the client-side OAuth
  settings, see [Client-side OAuth/OIDC authentication for Confluent components](co-authenticate-cp.md#co-authenticate-cp-client-oauth).

## Update connectors

To update a connector:

1. Edit the Connector CR configuration.

   If you change `metadata.name`, a new connector will be created.
2. Apply the changes using the `kubectl apply` command.

## Delete Connectors

To delete a connector, using the connector CR:

```bash
kubectl delete -f <connector CR>
```

To delete a connector, using the connector name:

```bash
kubectl delete connector <connector name>
```

## View state of connectors

To see the current status of a connector, run the following command:

```bash
kubectl describe connector <connector-CR-name>
```

The output contains the following fields:

`conditions`
: The latest observations of the connector.

`connectRestEndpoint`
: The rest endpoint of the Connect cluster.

`connectorState`
: The actual status of the connector instance.

`failedTasks`
: The connector tasks with state `FAILED`.

`failedTasksCount`
: The number of failed tasks.

`kafkaClusterId`
: The Kafka cluster Id the connector belongs to.

`restartPolicy`
: The policy to restart failed tasks of the connector.

`state`
: The CR state of the connector.

`tasksReady`
: The number of running tasks based on `taskMax`.

`trace`
: The error trace message for the connector instance.

`workerId`
: The worker Id for the connector instance.

You can run the following command to get the up-to-date list of the above status
fields:

```bash
kubectl explain connector.status
```

## Pause or restart connectors

You can use [Connector CR annotations](co-configure-misc.md#co-annotations) or [CFK CLI
plugins](co-install-plugin.md#co-install-plugin) tool to pause or restart the connectors that are
managed by CFK.

The following are the annotations to pause or restart connectors:

```bash
platform.confluent.io/pause-connector
platform.confluent.io/restart-connector
platform.confluent.io/restart-task
platform.confluent.io/resume-connector
```

Run the following command to add the annotation to pause, restart, resume a connector, or
to restart a connector task:

```bash
kubectl annotate connector <connector_CR_name> \
-n <namespace> \
<annotation>="<annotation value>"
```

For example, to pause a connector:

```bash
kubectl annotate connector my_connector_1 \
-n confluent \
platform.confluent.io/pause-connector="true"
```

you can install the [confluent kubectl plugins](co-install-plugin.md#co-install-plugin), and
use the [connector](co-plugin-cli/kubectl_confluent_connector.md#kubectl-confluent-connector) option to pause, restart,
resume a connector, or restart a connector task:

```bash
kubectl confluent connector pause [flags]
kubectl confluent connector restart [flags]
kubectl confluent connector restart-task [flags]
kubectl confluent connector resume [flags]
```