:_mod-docs-content-type: CONCEPT

// Module included in the following assemblies:
//
// assembly-config.adoc

[id='con-config-mirrormaker2-sync-{context}']
= Synchronizing consumer group offsets

[role="_abstract"]
Configure MirrorMaker 2 to synchronize consumer group offsets from the source cluster to the target cluster.

The `MirrorSourceConnector` and `MirrorCheckpointConnector` work together using internal topics to coordinate offset tracking between clusters.

`offset-syncs` topic:: Format: `mm2-offset-syncs<separator><cluster_alias><separator>internal` +
Populated by the `MirrorSourceConnector`, this topic stores offset mappings between source and target clusters. By default, it's created in the source cluster.
`checkpoints` topic:: 
Format: `<cluster_alias><separator>checkpoints<separator>internal` +
Populated by the `MirrorCheckpointConnector` in the target cluster, this topic captures the last committed offsets for each consumer group.

Configuring the `MirrorCheckpointConnector` to emit periodic offset checkpoints enables:

* Active/passive offset synchronization
* Failover recovery as consumers switch to the target cluster at the correct position

Offset synchronization occurs at regular intervals.
The `MirrorCheckpointConnector` emits checkpoints for all consumer groups enabled by the group configuration.
However, applying those checkpoints in the target cluster requires the target group to be inactive.
If consumers switch to the target cluster between checkpoints, they might reprocess some messages because offset synchronization is asynchronous and the last synchronized offset lags behind the source.
This duplication is expected.

NOTE: The connector continues to emit checkpoints to the checkpoints topic even if the target consumer group is active. 
However, the offsets cannot be committed while the consumer group has active members. 
In this case, the connector logs a warning and the offsets are not synchronized.

Each connector is configured separately.

.Example configuration to enable offset synchronization
[source,yaml,subs="+quotes,attributes"]
----
apiVersion: {KafkaMirrorMaker2ApiVersion}
kind: KafkaMirrorMaker2
metadata:
  name: my-mirror-maker2
  # ...
  mirrors:
    - source:
        alias: "my-cluster-source"
        bootstrapServers: my-cluster-source-kafka-bootstrap:9092
      sourceConnector:
        config:
          offset-syncs.topic.replication.factor: -1 # <1>
          refresh.topics.interval.seconds: 60 # <2>
      checkpointConnector:
        config:
          sync.group.offsets.enabled: true # <3>
          sync.group.offsets.interval.seconds: 60 # <4>
          emit.checkpoints.interval.seconds: 60 # <5>
          refresh.groups.interval.seconds: 600 # <6>
          checkpoints.topic.replication.factor: -1 # <7> 
----
<1> Replication factor for the `offset-syncs` internal topic that maps the offsets of the source and target clusters. 
A value of -1 uses the broker’s default replication factor, which is typically set to provide resilience (for example, 3).
<2> Optional setting to change the frequency of checks for new topics.
<3> Enables consumer group offset synchronization.
<4> The frequency of the synchronization.
<5> Adjusts the frequency of checks for offset tracking. If you change the frequency of offset synchronization, you might also need to adjust the frequency of these checks.  
<6> The frequency of checks for new consumer groups.
<7> Replication factor for the internal checkpoints topic that stores the last committed offsets for each consumer group. 
A value of -1 uses the broker’s default replication factor, which is typically set to provide resilience (for example, 3).

TIP: Java applications can use the `RemoteClusterUtils` utility to fetch remote offsets for a consumer group from the checkpoints topic.

== Changing the location of the offset synchronization topic

By default, the `offset-syncs` topic used by the `MirrorSourceConnector` is created in the source cluster.
You can change this behavior by setting the `offset-syncs.topic.location` connector configuration to `target`.

Placing the `offset-syncs` topic on the target cluster allows the `MirrorSourceConnector` to run when it has read-only access to the source cluster.
The connector must still have read and write access to the `offset-syncs` topic on the cluster where the topic is located.

== Listing the offsets of connectors

To list the offset positions of the internal MirrorMaker 2 connectors, use the same configuration that's used to manage Kafka Connect connectors.
For more information on setting up the configuration and listing offsets, see xref:proc-listing-connector-offsets-{context}[].

In this example, the `sourceConnector` configuration is updated to return the connector offset position.
The offset information is written to a specified `ConfigMap`.

.Example configuration for MirrorMaker 2 connector
[source,yaml,subs="+quotes,attributes"]
----
apiVersion: {KafkaMirrorMaker2ApiVersion}
kind: KafkaMirrorMaker2
metadata:
  name: my-mirror-maker2
spec:
  version: {DefaultKafkaVersion}
  # ...
  target:
    alias: "my-cluster-target"
    bootstrapServers: my-cluster-target-kafka-bootstrap:9092
    groupId: my-mirror-maker2-group
    configStorageTopic: my-mirror-maker2-config
    offsetStorageTopic: my-mirror-maker2-offset
    statusStorageTopic: my-mirror-maker2-status
  mirrors:
  - source:
      alias: "my-cluster-source"
      bootstrapServers: my-cluster-source-kafka-bootstrap:9092
    sourceConnector:
      listOffsets:
        toConfigMap:
          name: my-connector-offsets
        # ...    
----

You must apply the following annotations to the `KafkaMirrorMaker2` resource be able to manage connector offsets:

* `strimzi.io/connector-offsets`
* `strimzi.io/mirrormaker-connector`

The `strimzi.io/mirrormaker-connector` annotation must be set to the name of the connector.
These annotations remain until the operation succeeds or they are manually removed from the resource.

MirrorMaker 2 connectors are named using the aliases of the source and target clusters, followed by the connector type: `<source_alias>&#45;&#62;<target_alias>.<connector_type>`.

In the following example, the annotations are applied for a connector named `my-cluster-source&#45;&#62;my-cluster-target.MirrorSourceConnector`.

.Example application of annotations for connector
[source,shell]
----
kubectl annotate kafkamirrormaker2 my-mirror-maker-2 strimzi.io/connector-offsets=list strimzi.io/mirrormaker-connector="my-cluster-source->my-cluster-target.MirrorSourceConnector" -n kafka
----

The offsets are listed in the specified `ConfigMap`.
Strimzi puts the offset information into a `.json` property named after the connector. 
This does not overwrite any other properties when updating an existing `ConfigMap`.

.Example source connector offset list
[source,yaml,subs="+attributes"]
----
apiVersion: v1
kind: ConfigMap
metadata:
  # ...
  ownerReferences: # <1>
  - apiVersion: {KafkaMirrorMaker2ApiVersion}
    blockOwnerDeletion: false
    controller: false
    kind: KafkaMirrorMaker2
    name: my-mirror-maker2
    uid: 637e3be7-bd96-43ab-abde-c55b4c4550e0
data: 
  my-cluster-source--my-cluster-target.MirrorSourceConnector.json: |- # <2>
    {
      "offsets": [
        {
          "partition": {
            "cluster": "east-kafka",
            "partition": 0,
            "topic": "mirrormaker2-cluster-configs"
          },
          "offset": {
            "offset": 0
          }
        }
      ]
    }
----
<1> The owner reference pointing to the `KafkaMirrorMaker2` resource. 
To provide a custom owner reference, create the `ConfigMap` in advance and set the owner reference.
<2> The `.json` property uses the connector name. Since `&#45;&#62;` characters are not allowed in `ConfigMap` keys, `&#45;&#62;` is changed to `--` in the connector name.

NOTE: It is possible to use configuration to xref:proc-altering-connector-offsets-{context}[alter] or xref:proc-resetting-connector-offsets-{context}[reset] connector offsets. 
This should be done with caution.