// Module included in the following assemblies:
//
// assembly-kafka-connect-configuration.adoc

[id='ref-kafka-connect-configuration-{context}']
= Kafka Connect configuration

Kafka Connect is configured using the `config` property in `KafkaConnect.spec` and `KafkaConnectS2I.spec`.
This property contains the Kafka Connect configuration options as keys.
The values can be one of the following JSON types:

* String
* Number
* Boolean

You can specify and configure the options listed in the {ApacheKafkaConnectConfig} with the exception of those options that are managed directly by {ProductName}.
Specifically, configuration options with keys equal to or starting with one of the following strings are forbidden:

* `ssl.`
* `sasl.`
* `security.`
* `listeners`
* `plugin.path`
* `rest.`
* `bootstrap.servers`

When a forbidden option is present in the `config` property, it is ignored and a warning message is printed to the Custer Operator log file.
All other options are passed to Kafka Connect.

IMPORTANT: The Cluster Operator does not validate keys or values in the `config` object provided.
When an invalid configuration is provided, the Kafka Connect cluster might not start or might become unstable.
In this circumstance, fix the configuration in the `KafkaConnect.spec.config` or `KafkaConnectS2I.spec.config` object, then the Cluster Operator can roll out the new configuration to all Kafka Connect nodes.

Certain options have default values:

* `group.id` with default value `connect-cluster`
* `offset.storage.topic` with default value `connect-cluster-offsets`
* `config.storage.topic` with default value `connect-cluster-configs`
* `status.storage.topic` with default value `connect-cluster-status`
* `key.converter` with default value `org.apache.kafka.connect.json.JsonConverter`
* `value.converter` with default value `org.apache.kafka.connect.json.JsonConverter`

These options are automatically configured in case they are not present in the `KafkaConnect.spec.config` or `KafkaConnectS2I.spec.config` properties.

.Example Kafka Connect configuration
[source,yaml,subs="attributes+"]
----
apiVersion: {KafkaApiVersion}
kind: KafkaConnect
metadata:
  name: my-connect
spec:
  # ...
  config:
    group.id: my-connect-cluster
    offset.storage.topic: my-connect-cluster-offsets
    config.storage.topic: my-connect-cluster-configs
    status.storage.topic: my-connect-cluster-status
    key.converter: org.apache.kafka.connect.json.JsonConverter
    value.converter: org.apache.kafka.connect.json.JsonConverter
    key.converter.schemas.enable: true
    value.converter.schemas.enable: true
    config.storage.replication.factor: 3
    offset.storage.replication.factor: 3
    status.storage.replication.factor: 3
  # ...
----
