Configures a Kafka Connect cluster.

[id='property-kafka-connect-config-{context}']
=== `config`
Use the `config` properties to configure Kafka options as keys.

Standard Apache Kafka Connect configuration may be provided, restricted to those properties not managed directly by Strimzi.

Configuration options that cannot be configured relate to:

* Kafka cluster bootstrap address
* Security (Encryption, Authentication, and Authorization)
* Listener / REST interface configuration
* Plugin path configuration

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 Strimzi.
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 Cluster 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` 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` properties.

There are exceptions to the forbidden options.
You can use three allowed `ssl` configuration options for client connection using a specific _cipher suite_ for a TLS version.
A cipher suite combines algorithms for secure connection and data transfer.
You can also configure the `ssl.endpoint.identification.algorithm` property to enable or disable hostname verification.

.Example Kafka Connect configuration
[source,yaml,subs="attributes+"]
----
apiVersion: {KafkaConnectApiVersion}
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
    ssl.cipher.suites: "TLS_ECDHE_RSA_WITH_AES_256_GCM_SHA384"
    ssl.enabled.protocols: "TLSv1.2"
    ssl.protocol: "TLSv1.2"
    ssl.endpoint.identification.algorithm: HTTPS
  # ...
----

For client connection using a specific _cipher suite_ for a TLS version, you can xref:con-common-configuration-ssl-reference[configure allowed `ssl` properties].
You can also xref:con-common-configuration-ssl-reference[configure the `ssl.endpoint.identification.algorithm` property] to enable or disable hostname verification.

[id='property-kafka-connect-logging-{context}']
=== `logging`
Kafka Connect has its own configurable loggers:

* `connect.root.logger.level`
* `log4j.logger.org.reflections`

Further loggers are added depending on the Kafka Connect plugins running.

Use a curl request to get a complete list of Kafka Connect loggers running from any Kafka broker pod:

[source,curl,subs=attributes+]
----
curl -s http://<connect-cluster-name>-connect-api:8083/admin/loggers/
----

Kafka Connect uses the Apache `log4j` logger implementation.

Use the `logging` property to configure loggers and logger levels.

You can set the log levels by specifying the logger and level directly (inline) or use a custom (external) ConfigMap.
If a ConfigMap is used, you set `logging.valueFrom.configMapKeyRef.name` property to the name of the ConfigMap containing the external logging configuration. Inside the ConfigMap, the logging configuration is described using `log4j.properties`. Both `logging.valueFrom.configMapKeyRef.name` and `logging.valueFrom.configMapKeyRef.key` properties are mandatory. A ConfigMap using the exact logging configuration specified is created with the custom resource when the Cluster Operator is running, then recreated after each reconciliation. If you do not specify a custom ConfigMap, default logging settings are used. If a specific logger value is not set, upper-level logger settings are inherited for that logger.
For more information about log levels, see {ApacheLoggers}.

Here we see examples of `inline` and `external` logging.

.Inline logging
[source,yaml,subs="+quotes,attributes"]
----
apiVersion: {KafkaConnectApiVersion}
kind: KafkaConnect
spec:
  # ...
  logging:
    type: inline
    loggers:
      connect.root.logger.level: "INFO"
  # ...
----

.External logging
[source,yaml,subs="+quotes,attributes"]
----
apiVersion: {KafkaConnectApiVersion}
kind: KafkaConnect
spec:
  # ...
  logging:
    type: external
    valueFrom:
      configMapKeyRef:
        name: customConfigMap
        key: connect-logging.log4j
  # ...
----

Any available loggers that are not configured have their level set to `OFF`.

If Kafka Connect was deployed using the Cluster Operator,
changes to Kafka Connect logging levels are applied dynamically.

If you use external logging, a rolling update is triggered when logging appenders are changed.

.Garbage collector (GC)

Garbage collector logging can also be enabled (or disabled) using the xref:con-common-configuration-garbage-collection-reference[`jvmOptions` property].
