// 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

* See the _Deploying Strimzi_ guide for instructions on running a:
** link:{BookURLDeploying}#cluster-operator-str[Cluster Operator^]
** link:{BookURLDeploying}#deploying-kafka-cluster-str[Kafka cluster^]
* Source and target Kafka clusters must be 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
      ssl.cipher.suites: "TLS_ECDHE_RSA_WITH_AES_256_GCM_SHA384" <9>
      ssl.enabled.protocols: "TLSv1.2"
      ssl.protocol: "TLSv1.2"
      ssl.endpoint.identification.algorithm: HTTPS <10>
  producer:
    bootstrapServers: my-target-cluster-kafka-bootstrap:9092
    abortOnSendFailure: false <11>
    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
      ssl.cipher.suites: "TLS_ECDHE_RSA_WITH_AES_256_GCM_SHA384" <12>
      ssl.enabled.protocols: "TLSv1.2"
      ssl.protocol: "TLSv1.2"
      ssl.endpoint.identification.algorithm: HTTPS <13>
  whitelist: "my-topic|other-topic" <14>
  resources: <15>
    requests:
      cpu: "1"
      memory: 2Gi
    limits:
      cpu: "2"
      memory: 2Gi
  logging: <16>
    type: inline
    loggers:
      mirrormaker.root.logger: "INFO"
  readinessProbe: <17>
    initialDelaySeconds: 15
    timeoutSeconds: 5
  livenessProbe:
    initialDelaySeconds: 15
    timeoutSeconds: 5
  metrics: <18>
    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: <19>
    "-Xmx": "1g"
    "-Xms": "1g"
  image: my-org/my-image:latest <20>
  template: <21>
    pod:
      affinity:
        podAntiAffinity:
          requiredDuringSchedulingIgnoredDuringExecution:
            - labelSelector:
                matchExpressions:
                  - key: application
                    operator: In
                    values:
                      - postgresql
                      - mongodb
              topologyKey: "kubernetes.io/hostname"
    connectContainer: <22>
      env:
        - name: JAEGER_SERVICE_NAME
          value: my-jaeger-service
        - name: JAEGER_AGENT_HOST
          value: jaeger-agent-name
        - name: JAEGER_AGENT_PORT
          value: "6831"
  tracing: <23>
    type: jaeger
----
<1> xref:con-common-configuration-replicas-reference[The number of replica nodes].
<2> xref:con-common-configuration-bootstrap-reference[Bootstrap servers] for consumer and producer.
<3> xref:property-consumer-group-reference[Group ID for the consumer].
<4> xref:property-consumer-streams-reference[The number of consumer streams].
<5> xref:property-consumer-offset-autocommit-reference[The offset auto-commit interval in milliseconds].
<6> xref:con-common-configuration-trusted-certificates-reference[TLS encryption] with key names under which TLS certificates are stored in X.509 format for consumer or producer. If certificates are stored in the same secret, it can be listed multiple times.
<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 xref:property-consumer-config-reference[consumer] and xref:property-producer-config-reference[producer].
<9> xref:con-common-configuration-ssl-reference[SSL properties] for external listeners to run with a specific _cipher suite_ for a TLS version.
<10> xref:type-KafkaMirrorMakerConsumerSpec-reference[Hostname verification is enabled] by setting to `HTTPS`. An empty string disables the verification.
<11> If the xref:property-producer-abort-on-send-reference[`abortOnSendFailure` property] is set to `true`, Kafka MirrorMaker will exit and the container will restart following a send failure for a message.
<12> xref:con-common-configuration-ssl-reference[SSL properties] for external listeners to run with a specific _cipher suite_ for a TLS version.
<13> xref:type-KafkaMirrorMakerProducerSpec-reference[Hostname verification is enabled] by setting to `HTTPS`. An empty string disables the verification.
<14> A xref:property-mm-whitelist-reference[_whitelist_ of topics] mirrored from source to target Kafka cluster.
<15> Requests for reservation of xref:con-common-configuration-resources-reference[supported resources], currently `cpu` and `memory`, and limits to specify the maximum resources that can be consumed.
<16> Specified xref:property-mm-loggers-reference[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.
<17> xref:con-common-configuration-healthchecks-reference[Healthchecks] to know when to restart a container (liveness) and when a container can accept traffic (readiness).
<18> xref:con-common-configuration-prometheus-reference[Prometheus metrics], which are enabled with configuration for the Prometheus JMX exporter in this example. You can enable metrics without further configuration using `metrics: {}`.
<19> xref:con-common-configuration-jvm-reference[JVM configuration options] to optimize performance for the Virtual Machine (VM) running Kafka MirrorMaker.
<20> ADVANCED OPTION: xref:con-common-configuration-images-reference[Container image configuration], which is recommended only in special situations.
<21> xref:assembly-customizing-kubernetes-resources-str[Template customization]. Here a pod is scheduled with anti-affinity, so the pod is not scheduled on nodes with the same hostname.
<22> Environment variables are also xref:ref-tracing-environment-variables-str[set for distributed tracing using Jaeger].
<23> xref:assembly-distributed-tracing-str[Distributed tracing is enabled for Jaeger].
+
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>_
