How to not mirror 'heartbeats' topic

[scottc-git]

Hello,

Is there a way to tell MM2 to not mirror 'heartbeats' topic? MM2 (technically the schema transfer SMT) is throwing gazillion errors saying it could not find the matching avro schema from the source Kafka but I don't need to mirror them in the first place. I thought having needed topics in the topicsPattern will automatically exclude all the other (internal) topics, but since it did not work, I even added to topicsExcludePattern list, which I felt unnecessary. Still, it did not work.

2022-08-02 21:41:00,226 ERROR Error encountered in task mirrorref->strimzi.MirrorSourceConnector-1. Executing stage 'TRANSFORMATION' with class 'cricket.jmoore.kafka.connect.transforms.SchemaRegistryTransfer', where source record is = SourceRecord{sourcePartition={cluster=mirrorref, partition=0, topic=heartbeats}, sourceOffset={offset=8268}} ConnectRecord{topic='heartbeats', kafkaPartition=0, key=[B@46b30586, keySchema=Schema{BYTES}, value=[B@6dd4a232, valueSchema=Schema{BYTES}, timestamp=1607454781621, headers=ConnectHeaders(headers=)}. (org.apache.kafka.connect.runtime.errors.LogReporter) [task-thread-mirrorref->strimzi.MirrorSourceConnector-1]
org.apache.kafka.connect.errors.ConnectException: cricket.jmoore.confluent.kafka.schemaregistry.client.rest.exceptions.RestClientException: Schema 1 not found; error code: 40403
        at cricket.jmoore.kafka.connect.transforms.SchemaRegistryTransfer.copySchema(SchemaRegistryTransfer.java:219)
        at cricket.jmoore.kafka.connect.transforms.SchemaRegistryTransfer.apply(SchemaRegistryTransfer.java:179)
        at org.apache.kafka.connect.runtime.TransformationChain.lambda$apply$0(TransformationChain.java:50)
        at org.apache.kafka.connect.runtime.errors.RetryWithToleranceOperator.execAndRetry(RetryWithToleranceOperator.java:156)
        at org.apache.kafka.connect.runtime.errors.RetryWithToleranceOperator.execAndHandleError(RetryWithToleranceOperator.java:190)
        at org.apache.kafka.connect.runtime.errors.RetryWithToleranceOperator.execute(RetryWithToleranceOperator.java:132)
        at org.apache.kafka.connect.runtime.TransformationChain.apply(TransformationChain.java:50)
        at org.apache.kafka.connect.runtime.WorkerSourceTask.sendRecords(WorkerSourceTask.java:323)
        at org.apache.kafka.connect.runtime.WorkerSourceTask.execute(WorkerSourceTask.java:248)
        at org.apache.kafka.connect.runtime.WorkerTask.doRun(WorkerTask.java:182)
        at org.apache.kafka.connect.runtime.WorkerTask.run(WorkerTask.java:231)
        at java.base/java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:515)
        at java.base/java.util.concurrent.FutureTask.run(FutureTask.java:264)
        at java.base/java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1128)
        at java.base/java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:628)
        at java.base/java.lang.Thread.run(Thread.java:829)
Caused by: cricket.jmoore.confluent.kafka.schemaregistry.client.rest.exceptions.RestClientException: Schema 1 not found; error code: 40403
        at cricket.jmoore.confluent.kafka.schemaregistry.client.rest.RestService.sendHttpRequest(RestService.java:209)
        at cricket.jmoore.confluent.kafka.schemaregistry.client.rest.RestService.httpRequest(RestService.java:235)
        at cricket.jmoore.confluent.kafka.schemaregistry.client.rest.RestService.getId(RestService.java:415)
        at cricket.jmoore.confluent.kafka.schemaregistry.client.rest.RestService.getId(RestService.java:408)
        at cricket.jmoore.confluent.kafka.schemaregistry.client.CachedSchemaRegistryClient.getSchemaByIdFromRegistry(CachedSchemaRegistryClient.java:123)
        at cricket.jmoore.confluent.kafka.schemaregistry.client.CachedSchemaRegistryClient.getBySubjectAndId(CachedSchemaRegistryClient.java:190)
        at cricket.jmoore.confluent.kafka.schemaregistry.client.CachedSchemaRegistryClient.getById(CachedSchemaRegistryClient.java:169)
        at cricket.jmoore.kafka.connect.transforms.SchemaRegistryTransfer.copySchema(SchemaRegistryTransfer.java:216)
        ... 15 more

Did someone run into this? Here's my MM2 spec (shortened version):

apiVersion: kafka.strimzi.io/v1beta2
kind: KafkaMirrorMaker2
metadata:
  name: strimzi
spec:
  version: 2.8.0
  replicas: 1
  # Custom Kafka Connect image with additional plugins.
  # https://strimzi.io/docs/master/#creating-new-image-from-base-str
  # https://example.com/tdp/image/strimzi-kafka-connect-extended
  image: example.com/k8s/dev:strimzi-kafka-connect-extended-strimzi-0-27-1-32838c44-62188

  clusters:
    - alias: sourcekafka1
    - alias: sourcekafka2
    # Strimzi Kafka cluster to mirror to.
    - alias: strimzi
      bootstrapServers: strimzi-kafka-bootstrap:9093
  connectCluster: strimzi

  externalConfiguration:
    env:
      - name: INPUT_MIRROR_START_TIME
        valueFrom:
          configMapKeyRef:
            name: input-mirror-start-time-config
            key: INPUT_MIRROR_START_TIME

  mirrors:
    # - Mirror required Reference Data.
    - sourceCluster: sourcekafka1
      targetCluster: strimzi
      topicsPattern: ^master-data.(foo.v2|bar.v1)$
      topicsExcludePattern: >-
        __consumer_offsets,
        _schemas,
        heatbeats
      groupsPattern: "do-not-mirror-groups"
      sourceConnector:
        tasksMax: 10
        config:
          replication.factor: 1
          offset-syncs.topic.replication.factor: 1
          sync.topic.acls.enabled: "false"
          refresh.topics.interval.seconds: 60
          replication.policy.separator: ""
          replication.policy.class: "io.strimzi.kafka.connect.mirror.IdentityReplicationPolicy"

          transforms: AvroSchemaTransfer
          # Copy Avro schemas.
          # https://www.confluent.io/blog/kafka-connect-tutorial-transfer-avro-schemas-across-schema-registry-clusters/
          transforms.AvroSchemaTransfer.type: cricket.jmoore.kafka.connect.transforms.SchemaRegistryTransfer
          transforms.AvroSchemaTransfer.transfer.message.keys: "false"
          transforms.AvroSchemaTransfer.src.schema.registry.url: http://override.me:8081
          transforms.AvroSchemaTransfer.dest.schema.registry.url: http://schema-registry-client:8081
          transforms.AvroSchemaTransfer.schema.capacity: 5000

          # Enable debugging
          # https://cwiki.apache.org/confluence/display/KAFKA/KIP-298%3A+Error+Handling+in+Connect
          errors.log.enable: "true"
          errors.log.include.messages: "true"

          # Skip bad messages, usually due to missing schemas
          errors.tolerance: all
      heartbeatConnector:
        config:
          heartbeats.topic.replication.factor: 1
      checkpointConnector:
        config:
          checkpoints.topic.replication.factor: 1
          refresh.groups.interval.seconds: 600
          sync.group.offsets.enabled: true
          sync.group.offsets.interval.seconds: 60
          emit.checkpoints.interval.seconds: 60
          replication.policy.class: "io.strimzi.kafka.connect.mirror.IdentityReplicationPolicy"

    # - Mirror transactional data
    - sourceCluster: sourcekafka2
      targetCluster: strimzi
      topicsPattern: >-
        txn-data-foo-v2,
        txn-data-bar-v2,
        txn-data-goo-v2,
      topicsExcludePattern: >-
        __consumer_offsets,
        _schemas,
        heatbeats
      groupsPattern: "do-not-mirror-groups"
      sourceConnector:
        tasksMax: 10
        config:
          replication.factor: 1
          offset-syncs.topic.replication.factor: 1
          sync.topic.acls.enabled: "false"
          refresh.topics.interval.seconds: 60
          replication.policy.separator: ""
          replication.policy.class: "io.strimzi.kafka.connect.mirror.IdentityReplicationPolicy"

          transforms: FilterOldMessages, AvroSchemaTransfer

          # Don't mirror old messages.
          transforms.FilterOldMessages.type: example.library.kafka.ConnectFilterTimeRange
          transforms.FilterOldMessages.start.time: ${env:INPUT_MIRROR_START_TIME}

          # Copy Avro schemas.
          # https://www.confluent.io/blog/kafka-connect-tutorial-transfer-avro-schemas-across-schema-registry-clusters/
          transforms.AvroSchemaTransfer.type: cricket.jmoore.kafka.connect.transforms.SchemaRegistryTransfer
          transforms.AvroSchemaTransfer.transfer.message.keys: "false"
          transforms.AvroSchemaTransfer.src.schema.registry.url: http://override.me:8081
          transforms.AvroSchemaTransfer.dest.schema.registry.url: http://schema-registry-client:8081
          transforms.AvroSchemaTransfer.schema.capacity: 5000

          # Enable debugging
          errors.log.enable: "true"
          errors.log.include.messages: "true"

          # Skip bad messages, usually due to missing schemas
          errors.tolerance: all
      heartbeatConnector:
        config:
          heartbeats.topic.replication.factor: 1
      checkpointConnector:
        config:
          checkpoints.topic.replication.factor: 1
          refresh.groups.interval.seconds: 600
          sync.group.offsets.enabled: true
          sync.group.offsets.interval.seconds: 60
          emit.checkpoints.interval.seconds: 60
          replication.policy.class: "io.strimzi.kafka.connect.mirror.IdentityReplicationPolicy"
  logging:
    type: inline
    loggers:
      log4j.logger.org.apache.kafka.connect.mirror.MirrorSourceTask: ERROR

ENV
Strimzi: 0.27.1
Kafka: 2.8.0

[scholzj]

If you remove the heartbeatConnector section, you should basically disable the heartbeat connector. Would that help? I think the heartbeat connector expects to be mirrored. So disabling should be the right way?

[scottc-git]

Thanks for the reply @scholzj I deleted & redeployed the kmm2 with the heartbeatConnect section removed, but it's still printing the errors. Hmm....

[scholzj]

Did you also tried to just delete the topic? It might be processing some old messages maybe?

[scottc-git]

I believe you mean the heartbeats topic in source cluster of which I am not the owner. I can certainly delete in my cluster but the errors are about the messages in the source topic. @scholzj

[mdraevich]

@scholzj Hi!
Sorry for replying to an old thread :)

Seems like the question was about this behavior.

HeartbeatConnector does emit heartbeat messages to a heartbeats topic of the target Kafka cluster.

But MirrorSourceConnector appears to replicate the heartbeats topic from the source Kafka cluster to the target cluster by default.
Due to this behavior, we observe some WARN messages about ACL violations in the source Kafka cluster, as MirrorSourceConnector attempts to read the heartbeats topic for replication.

What do you think? I couldn't find the heartbeats.replication.enabled option described in KIP-1089 in Strimzi 0.45.0.
Have a nice day!

---

UPD: seems like the feature I asked is released in 4.0.0 (jira task).
The latest Strimzi (0.45.0) does support 3.9.0. That means, the feature I asked cannot be configured in the latest Strimzi (0.45.0).