[CRD-v1] Redesign for KafkaMirrorMaker2 CRD

[katheris]

Related problem

The current version of the KafkaMirrorMaker2 CRD requires the following example CR:

kind: KafkaMirrorMaker2
metadata:
  name: my-mirror-maker-2
spec:
  version: 4.0.0
  replicas: 1
  connectCluster: "cluster-b"
  clusters:
  - alias: "cluster-a"
    bootstrapServers: cluster-a-kafka-bootstrap:9092
  - alias: "cluster-b"
    bootstrapServers: cluster-b-kafka-bootstrap:9092
    config:
      config.storage.replication.factor: -1
      offset.storage.replication.factor: -1
      status.storage.replication.factor: -1
  mirrors:
  - sourceCluster: "cluster-a"
    targetCluster: "cluster-b"
    sourceConnector:
      tasksMax: 1
      config:
        replication.factor: -1
        offset-syncs.topic.replication.factor: -1
        sync.topic.acls.enabled: "false"
        refresh.topics.interval.seconds: 600
    checkpointConnector:
      tasksMax: 1
      config:
        checkpoints.topic.replication.factor: -1
        sync.group.offsets.enabled: "false"
        refresh.groups.interval.seconds: 600
    topicsPattern: ".*"
    groupsPattern: ".*"

This will deploy a Kafka Connect cluster that is storing it's data in cluster-b and deploy the MirrorSourceConnector and MirrorCheckpointConnector mirroring from cluster-a to cluster-b.

Currently the names for the config, offset and status topics are not required, but if the user does want to set them they must be set under the cluster that is listed as the targetCluster in each mirror. Strimzi also enforces that the connectCluster property be set to the same as the targetCluster of each mirror. This is not strictly required by Kafka but is generally recommended for at least the MirrorSourceConnector and MirrorCheckpointConnector.

Although in theory the API allows the user to specify many different clusters and many different mirroring routes in the same file, the reality is that since the target must match connectCluster all routes must replicate to the same Kafka cluster.

Based on chatting to users of Strimzi the KafkaMirrorMaker2 CR doesn't seem the easiest for users. The introduction of the v1 API gives us the opportunity to change the KafkaMirrorMaker2 CRD in a way to make it more intuitive.

Suggested solution

I propose the following CR:

apiVersion: kafka.strimzi.io/v1beta2
kind: KafkaMirrorMaker2
metadata:
  name: my-mirror-maker-2
spec:
  version: 4.0.0
  replicas: 1
  targetCluster:
    alias: "cluster-c"
    bootstrapServers: cluster-c-kafka-bootstrap:9092
    config:
      config.storage.topic: my-mirror-maker-config
      config.storage.replication.factor: -1
      offset.storage.topic: my-mirror-maker-offset
      offset.storage.replication.factor: -1
      status.storage.topic: my-mirror-maker-status
      status.storage.replication.factor: -1
  sourceClusters:
  - alias: "cluster-a"
    bootstrapServers: cluster-a-kafka-bootstrap:9092
  - alias: "cluster-b"
    bootstrapServers: cluster-b-kafka-bootstrap:9092
  mirrors:
  - sourceCluster: "cluster-a"
    sourceConnector:
      tasksMax: 1
      config:
        replication.factor: -1
        offset-syncs.topic.replication.factor: -1
        sync.topic.acls.enabled: "false"
        refresh.topics.interval.seconds: 600
    checkpointConnector:
      tasksMax: 1
      config:
        checkpoints.topic.replication.factor: -1
        sync.group.offsets.enabled: "false"
        refresh.groups.interval.seconds: 600
    topicsPattern: ".*"
    groupsPattern: ".*"
  - sourceCluster: "cluster-b"
    sourceConnector:
      tasksMax: 1
      config:
        replication.factor: -1
        offset-syncs.topic.replication.factor: -1
        sync.topic.acls.enabled: "false"
        refresh.topics.interval.seconds: 600
    checkpointConnector:
      tasksMax: 1
      config:
        checkpoints.topic.replication.factor: -1
        sync.group.offsets.enabled: "false"
        refresh.groups.interval.seconds: 600
    topicsPattern: ".*"
    groupsPattern: ".*"

So the key changes are:

• The user specifies a single targetCluster in their CR, and this cluster is used for configuring the storage of the underlying Connect cluster
• The user is required to specify the topic names for the targetCluster
• The user specifies a set of sourceClusters rather than generic clusters
• For each mirror the user only specifies the sourceCluster, since the targetCluster is set at the CR level
• As before the connector names are generated as <SOURCE_CLUSTER_ALIAS>-><TARGET_CLUSTER_ALIAS>, e.g. cluster-a->cluster-c

This API better guides users to deploy MirrorMaker2 in the recommended way. If a user wants to deploy a more complex or non-recommended topology they can always use the KafkaConnect and KafkaConnector CRs directly.

MirrorHeartbeatConnector

The MirrorHeartbeatConnector is confusing to configure using the existing CR. For a MirrorHeartbeatConnector that is related to a MirrorSourceConnector replicating from cluster-a to cluster-c, the underlying Connect cluster should be associated with the source cluster (cluster-a), not the target (cluster-c) (since that is where it produces messages). However the source.alias and target.alias must be set as cluster-a and cluster-c respectively, otherwise the contents of the messages in the heartbeat topic don't make sense (they include these aliases). With the existing CR this means when using the MirrorHeartbeatConnector the user must specify connection details for the source cluster under the target cluster alias (see this comment for an example #11695 (comment)).

Since the MirrorHeartbeatConnector is not that commonly used and is confusing to configure we have a few options to make the situation better:

1. Removing it entirely from the KafkaMirrorMaker2 CRD, and instead provide an example file for how to configure it using the standard KafkaConnect and KafkaConnector CRs
2. Providing a new KafkaMirrorMaker2Heartbeat CR that is tailored for the MirrorHeartbeatConnector
3. Update the heartbeatConnector section of the KafkaMirrorMaker2 CR to allow the user to specify the Connect topics and other Connect cluster configs directly there and having Strimzi create a second Connect cluster connected to the source Kafka cluster if the heartbeatConnector is configured.
4. Require the user to specify the Connect topics and other Connect cluster for the sourceCluster under the sourceClusters section when the heartbeatConnector is configured.

Given the fact that the MirrorHeartbeatConnector is not so commonly used, I would propose option 1.

Alternatives

No response

Additional context

No response

[katheris]

@mimaison @scholzj FYI

[scholzj]

At a quick look, I think the changes suggested in the first part should be feasible. The Heartbeat part is a bit harder. I think the options 1 and 2 are not really feasible. For 3 and 4, I do not think I really understand what exactly is it expected to do.

[katheris]

I'm guessing the reason that 1 and 2 would not be feasible is in terms of upgrade (let me know if there's a different reason). For most of the v1 changes we would expect automatic upgrade to be possible and obviously upgrade with no downtime. But for the MirrorHeartbeatConnector since it is a mechanism for monitoring the health of the MirrorSourceConnector and lag time of replication I think it would be acceptable for there to be downtime as part of the upgrade. So from that PoV I think it is feasible, but does have the drawback that users would have to manually update their deployments first.

In terms of options 3 and 4, this is how they could look..

Option 3:

apiVersion: kafka.strimzi.io/v1beta2
kind: KafkaMirrorMaker2
metadata:
  name: my-mirror-maker-2
spec:
  version: 4.0.0
  replicas: 1
  targetCluster:
    alias: "cluster-c"
    bootstrapServers: cluster-c-kafka-bootstrap:9092
    config:
      config.storage.topic: my-mirror-maker-config
      config.storage.replication.factor: -1
      offset.storage.topic: my-mirror-maker-offset
      offset.storage.replication.factor: -1
      status.storage.topic: my-mirror-maker-status
      status.storage.replication.factor: -1
  sourceClusters:
  - alias: "cluster-a"
    bootstrapServers: cluster-a-kafka-bootstrap:9092
  mirrors:
  - sourceCluster: "cluster-a"
    sourceConnector:
      tasksMax: 1
      config:
        replication.factor: -1
        offset-syncs.topic.replication.factor: -1
        sync.topic.acls.enabled: "false"
        refresh.topics.interval.seconds: 600
    checkpointConnector:
      tasksMax: 1
      config:
        checkpoints.topic.replication.factor: -1
        sync.group.offsets.enabled: "false"
        refresh.groups.interval.seconds: 600
    topicsPattern: ".*"
    groupsPattern: ".*"
    heartbeatConnector:
      config:
        heartbeats.topic.replication.factor: -1
        emit.heartbeats.interval.seconds: 10
        config.storage.topic: my-mirror-maker-config
        config.storage.replication.factor: -1
        offset.storage.topic: my-mirror-maker-offset
        offset.storage.replication.factor: -1
        status.storage.topic: my-mirror-maker-status
        status.storage.replication.factor: -1

Option 4:

apiVersion: kafka.strimzi.io/v1beta2
kind: KafkaMirrorMaker2
metadata:
  name: my-mirror-maker-2
spec:
  version: 4.0.0
  replicas: 1
  targetCluster:
    alias: "cluster-c"
    bootstrapServers: cluster-c-kafka-bootstrap:9092
    config:
      config.storage.topic: my-mirror-maker-config
      config.storage.replication.factor: -1
      offset.storage.topic: my-mirror-maker-offset
      offset.storage.replication.factor: -1
      status.storage.topic: my-mirror-maker-status
      status.storage.replication.factor: -1
  sourceClusters:
  - alias: "cluster-a"
    bootstrapServers: cluster-a-kafka-bootstrap:9092
    config:
        config.storage.topic: my-mirror-maker-config
        config.storage.replication.factor: -1
        offset.storage.topic: my-mirror-maker-offset
        offset.storage.replication.factor: -1
        status.storage.topic: my-mirror-maker-status
        status.storage.replication.factor: -1
  mirrors:
  - sourceCluster: "cluster-a"
    sourceConnector:
      tasksMax: 1
      config:
        replication.factor: -1
        offset-syncs.topic.replication.factor: -1
        sync.topic.acls.enabled: "false"
        refresh.topics.interval.seconds: 600
    checkpointConnector:
      tasksMax: 1
      config:
        checkpoints.topic.replication.factor: -1
        sync.group.offsets.enabled: "false"
        refresh.groups.interval.seconds: 600
    topicsPattern: ".*"
    groupsPattern: ".*"
    heartbeatConnector:
      config:
        heartbeats.topic.replication.factor: -1
        emit.heartbeats.interval.seconds: 10

In both setups the outcome is a Connect cluster storing state in cluster-c which is running the MirrorSourceConnector and MirrorCheckpointConnector, and then also a Connect cluster storing state in cluster-a which is running the MirrorHeartbeatConnector. It means you can specify everything in one go, but introduces the confusion of having two Connect clusters for a single CR, and potentially requires you to change where the replicas field is placed.

[katheris]

For the MirrorHeartbeatConnector the fundamental problem we have is that it should be deployed in a Connect cluster storing state in the source cluster, and should be configured to produce to the source cluster, but the source and target aliases need to match the aliases provided to the MirrorSourceConnector and MirrorCheckpointConnector otherwise the contents of the records in the heartbeats topic makes no sense. So it needs to be handled quite differently to the other two, and since today Strimzi handles them all the same it's going to be hard to change it in a way that allows you to easily upgrade the CR.

[scholzj]

Not feasible was maybe not the right word. Sorry.

• The option 1 would of course work and is pretty simple to do. We just deprecate the heartbeat connector field from KafkaMirrorMaker2 and remove it in v1. And havng an example how to run a separate Connect with it is pretty simple. But is there actually anyone who wants to run a separate Connect cluster for the heartbeat connector? I doubt that.
  * The option 2 has not really much to do with v1 as we can't really convert anything into a new CRD. I would also argue that that is a loooot of effort for what you get.
• The options 3 and 4 seem like a complete hell in terms of implementation. You would be running two Connect clusters under one resource. That will be pretty complicated for error handling, marking it as ready.

---

To be honest, I have not met any users who understood the heartbeat connector and what it is supposed to do. And clearly I did not understand it myself either. And I probably still don't understand from how you describe it why is it there and why was it designed the way it was. So option 1 seems to me to be the easiest way to go from a pure Strimzi perspective. I doubt many people will use it, but I think they don't really use it today either (especially given it does not work).

But I do not get the design in the first place. Why do the mirror and checkpoint connectors need to run against Connect linked to target while the Heartbeat needs to run against Connect linked to source? I do not think the Heartbeat connector has any advanced logic and just sends periodically some message, or? You can do that pretty easily from a connector running against target as well, or? You would just have a sink connector that, based on a timer, sends messages with Kafka producer to a remote cluster 🤷‍♂️. Or maybe I have no idea what it is doing in the first place? 🤔

[mimaison]

I do not think the Heartbeat connector has any advanced logic and just sends periodically some message, or?

That's exactly it. It does not do anything else. But you need to send the heartbeats to the source cluster while the other 2 connectors send their data to the target cluster.

My experience with the heartbeat connector is that it brings little value, often much less than people expects. The complications it brings, both in terms of setup and reasoning, make it in most cases just not worth deploying.

[scholzj]

So, given that, should we go with the option 1? Remove the heartbeat connector from the KafkaMirrorMaker2 and let users who really want it deploy with as a completely Connect cluster + Connector? While that is not a great workaround, it seems there is no big need and demand for it. And it does not have any huge impact on our codebase. WDYT @katheris?

[scholzj]

@katheris Also please be aware of the latest updated about the enforcement of the Connect topics - #10075 (comment) - we will likely need to mirror that here as well.

[katheris]

@scholzj yeah my preference is for option 1.

What is the right way to get this properly agreed in the community, do you want a separate proposal or should we incorporate it in the existing v1 API proposal?

[scholzj]

What is the right way to get this properly agreed in the community, do you want a separate proposal or should we incorporate it in the existing v1 API proposal?

I think it should go through a proposal. Maybe I can add it to the one about the Connect and its topics? I might also need to do some prototyping to see that the new KafkaMirrorMaker2 structure works.

[katheris]

@scholzj yeah I'm happy for you to add it there, or if you want me to create a dedicated proposal for it then let me know.

On the MirrorHeartbeatConnector I was chatting more to @mimaison and an option I didn't consider is deploying the MirrorHeartbeatConnector to the target Connect cluster. So we could keep the heartbeatConnector field and when set Strimzi runs the MirrorHeartbeatConnector on the Connect cluster storing state in the target cluster, but updates the correct configuration to mean it is actually producing messages to the source cluster. We would probably need a warning in the docs though to tell users that if they are using MirrorHeartbeatConnector to track latency in mirroring it won't be accurate, since the connector will be writing to a "far away" Kafka cluster, and so that could introduce extra latency.