// Module included in the following assemblies:
//
// assembly-deployment-configuration-kafka-mirror-maker.adoc

[id='configuring-kafka-mirror-maker-{context}']
= Configuring Kafka MirrorMaker

Use the properties of the `KafkaMirrorMaker` resource to configure your Kafka MirrorMaker deployment.

You can configure access control for producers and consumers using TLS or SASL authentication.
This procedure shows a configuration that uses TLS encryption and authentication on the consumer and producer side.

.Prerequisites

* xref:cluster-operator-str[{ProductName} and Kafka is deployed]
* Source and target Kafka clusters are available

.Procedure

. Edit the `spec` properties for the `KafkaMirrorMaker` resource.
+
The properties you can configure are shown in this example configuration:
+
[source,yaml,subs="+quotes,attributes"]
----
apiVersion: {KafkaApiVersion}
kind: KafkaMirrorMaker
metadata:
  name: my-mirror-maker
spec:
  replicas: 3 <1>
  consumer:
    bootstrapServers: my-source-cluster-kafka-bootstrap:9092 <2>
    groupId: "my-group" <3>
    numStreams: 2 <4>
    offsetCommitInterval: 120000 <5>
    tls: <6>
      trustedCertificates:
      - secretName: my-source-cluster-ca-cert
        certificate: ca.crt
    authentication: <7>
      type: tls
      certificateAndKey:
        secretName: my-source-secret
        certificate: public.crt
        key: private.key
    config: <8>
      max.poll.records: 100
      receive.buffer.bytes: 32768
  producer:
    bootstrapServers: my-target-cluster-kafka-bootstrap:9092
    abortOnSendFailure: false <9>
    tls:
      trustedCertificates:
      - secretName: my-target-cluster-ca-cert
        certificate: ca.crt
    authentication:
      type: tls
      certificateAndKey:
        secretName: my-target-secret
        certificate: public.crt
        key: private.key
    config:
      compression.type: gzip
      batch.size: 8192
  whitelist: "my-topic|other-topic" <10>
  resources: <11>
    requests:
      cpu: "1"
      memory: 2Gi
    limits:
      cpu: "2"
      memory: 2Gi
  logging: <12>
    type: inline
    loggers:
      mirrormaker.root.logger: "INFO"
  readinessProbe: <13>
    initialDelaySeconds: 15
    timeoutSeconds: 5
  livenessProbe:
    initialDelaySeconds: 15
    timeoutSeconds: 5
  metrics: <14>
    lowercaseOutputName: true
    rules:
      - pattern: "kafka.server<type=(.+), name=(.+)PerSec\\w*><>Count"
        name: "kafka_server_$1_$2_total"
      - pattern: "kafka.server<type=(.+), name=(.+)PerSec\\w*,
        topic=(.+)><>Count"
        name: "kafka_server_$1_$2_total"
        labels:
          topic: "$3"
  jvmOptions: <15>
    "-Xmx": "1g"
    "-Xms": "1g"
  image: my-org/my-image:latest <16>
  template: <17>
      pod:
        affinity:
          podAntiAffinity:
            requiredDuringSchedulingIgnoredDuringExecution:
              - labelSelector:
                  matchExpressions:
                    - key: application
                      operator: In
                      values:
                        - postgresql
                        - mongodb
                topologyKey: "kubernetes.io/hostname"
----
+
<1> The number of replica nodes.
<2> Bootstrap servers for consumer and producer.
<3> Group ID for the consumer.
<4> The number of consumer streams.
<5> The offset auto-commit interval in milliseconds.
<6> TLS encryption with key names under which TLS certificates are stored in X.509 format for consumer or producer. For more details see xref:type-KafkaMirrorMakerTls-reference[`KafkaMirrorMakerTls` schema reference].
<7> Authentication for consumer or producer, using the xref:type-KafkaClientAuthenticationTls-reference[TLS mechanism], as shown here, using xref:type-KafkaClientAuthenticationOAuth-reference[OAuth bearer tokens], or a SASL-based xref:type-KafkaClientAuthenticationScramSha512-reference[SCRAM-SHA-512] or xref:type-KafkaClientAuthenticationPlain-reference[PLAIN] mechanism.
<8> Kafka configuration options for consumer and producer.
<9> If set to `true`, Kafka MirrorMaker will exit and the container will restart following a send failure for a message.
<10> Topics mirrored from source to target Kafka cluster.
<11> Requests for reservation of supported resources, currently `cpu` and `memory`, and limits to specify the maximum resources that can be consumed.
<12> Specified loggers and log levels added directly (`inline`) or indirectly (`external`) through a ConfigMap. A custom ConfigMap must be placed under the `log4j.properties` or `log4j2.properties` key. MirrorMaker has a single logger called `mirrormaker.root.logger`. You can set the log level to INFO, ERROR, WARN, TRACE, DEBUG, FATAL or OFF.
<13> Healthchecks to know when to restart a container (liveness) and when a container can accept traffic (readiness).
<14> Prometheus metrics, which are enabled with configuration for the Prometheus JMX exporter in this example. You can enable metrics without further configuration using `metrics: {}`.
<15> JVM configuration options to optimize performance for the Virtual Machine (VM) running Kafka MirrorMaker.
<16> ADVANCED OPTION: Container image configuration, which is xref:con-configuring-container-images-{context}[recommended only in special situations].
<17> xref:assembly-customizing-deployments-str[Template customization]. Here a pod is scheduled based with anti-affinity, so the pod is not scheduled on nodes with the same hostname.
+
WARNING: With the `abortOnSendFailure` property set to `false`, the producer attempts to send the next message in a topic. The original message might be lost, as there is no attempt to resend a failed message.

. Create or update the resource:
+
[source,shell,subs=+quotes]
kubectl apply -f _<your-file>_
