[Question] KafkaMirrorMaker2 2.7.0 compatiblity

[Joseph94m]

Hey! Apparently kafkaMirrormaker2 v2.7.0 was set to be released last Friday (November 6).

IMO this release is extremely important as it allows the replication of group offsets (something that wasn't possible pre 2.7.0 unless you compiled the trunk which can be quite messy and i can attest to that)

Today i've tried setting the version of my KafkaMirrorMaker2 CRD to 2.7.0 and deployed it on my target cluster.

apiVersion: kafka.strimzi.io/v1alpha1
kind: KafkaMirrorMaker2
metadata:
  name: my-mirror-maker-2
spec:
  version: 2.7.0
  replicas: 1
  clusters:
  - alias: "my-source-cluster"
    bootstrapServers: some-address:9094
  - alias: "my-target-cluster"
    bootstrapServers: kafka-dest-kafka-bootstrap.default.svc.cluster.local:9092
.......

I am using v0.20.0 for strimzi and co:

watchNamespaces: []
watchAnyNamespace: false
image:
  repository: strimzi
  name: operator
  tag: 0.20.0
logVolume: co-config-volume
logConfigMap: strimzi-cluster-operator
logLevel: ${env:STRIMZI_LOG_LEVEL:-INFO}
fullReconciliationIntervalMs: 120000
operationTimeoutMs: 300000
kubernetesServiceDnsDomain: cluster.local
tolerations: []
affinity: {}
nodeSelector: {}
zookeeper:
  image:
    repository: strimzi
    name: kafka
    tagPrefix: 0.20.0
kafka:
  image:
    repository: strimzi
    name: kafka
    tagPrefix: 0.20.0
kafkaConnect:
  image:
    repository: strimzi
    name: kafka
    tagPrefix: 0.20.0
kafkaConnects2i:
  image:
    repository: strimzi
    name: kafka
    tagPrefix: 0.20.0
topicOperator:
  image:
    repository: strimzi
    name: operator
    tag: 0.20.0
userOperator:
  image:
    repository: strimzi
    name: operator
    tag: 0.20.0
kafkaInit:
  image:
    repository: strimzi
    name: operator
    tag: 0.20.0
tlsSidecarEntityOperator:
  image:
    repository: strimzi
    name: kafka
    tagPrefix: 0.20.0
kafkaMirrorMaker:
  image:
    repository: strimzi
    name: kafka
    tagPrefix: 0.20.0
kafkaBridge:
  image:
    repository: strimzi
    name: kafka-bridge
    tag: 0.20.0
kafkaExporter:
  image:
    repository: strimzi
    name: kafka
    tagPrefix: 0.20.0
jmxTrans:
  image:
    repository: strimzi
    name: jmxtrans
    tag: 0.20.0
kafkaMirrorMaker2:
  image:
    repository: strimzi
    name: kafka
    tagPrefix: 0.20.0
cruiseControl:
  image:
    repository: strimzi
    name: kafka
    tagPrefix: 0.20.0
tlsSidecarCruiseControl:
  image:
    repository: strimzi
    name: kafka
    tagPrefix: 0.20.0
resources:
  limits:
    memory: 384Mi
    cpu: 1000m
  requests:
    memory: 384Mi
    cpu: 200m
livenessProbe:
  initialDelaySeconds: 10
  periodSeconds: 30
readinessProbe:
  initialDelaySeconds: 10
  periodSeconds: 30
imageRepositoryOverride: ""
imageTagOverride: ""
createGlobalResources: true

Strimzi-cluster-operator seems to be struggling with that and claims that 2.7.0 isn't in the list of supported versions.

│ 2020-11-09 08:51:01 ERROR AbstractOperator:238 - Reconciliation #20(watch) KafkaMirrorMaker2(default/my-mirror-maker-2): createOrUpdate failed     │
│ io.strimzi.operator.cluster.KafkaUpgradeException: Unsupported Kafka.spec.kafka.version: 2.7.0. Supported versions are: [2.5.1, 2.6.0, 2.5.0]      │
│     at io.strimzi.operator.cluster.model.KafkaVersion$Lookup.version(KafkaVersion.java:132) ~[io.strimzi.cluster-operator-0.20.0.jar:0.20.0]       │
│     at io.strimzi.operator.cluster.model.KafkaMirrorMaker2Cluster.fromCrd(KafkaMirrorMaker2Cluster.java:82) ~[io.strimzi.cluster-operator-0.20.0.j │
│     at io.strimzi.operator.cluster.operator.assembly.KafkaMirrorMaker2AssemblyOperator.createOrUpdate(KafkaMirrorMaker2AssemblyOperator.java:127)  │
│     at io.strimzi.operator.cluster.operator.assembly.KafkaMirrorMaker2AssemblyOperator.createOrUpdate(KafkaMirrorMaker2AssemblyOperator.java:74) ~ │
│     at io.strimzi.operator.common.AbstractOperator.lambda$reconcile$6(AbstractOperator.java:219) ~[io.strimzi.operator-common-0.20.0.jar:0.20.0]   │
│     at io.strimzi.operator.common.AbstractOperator.lambda$withLock$13(AbstractOperator.java:360) ~[io.strimzi.operator-common-0.20.0.jar:0.20.0]   │
│     at io.vertx.core.shareddata.impl.LocalAsyncLocks$LockWaiter.lambda$acquireLock$1(LocalAsyncLocks.java:65) ~[io.vertx.vertx-core-3.9.1.jar:3.9. │
│     at io.vertx.core.impl.ContextImpl.executeTask(ContextImpl.java:366) ~[io.vertx.vertx-core-3.9.1.jar:3.9.1]                                     │
│     at io.vertx.core.impl.EventLoopContext.lambda$executeAsync$0(EventLoopContext.java:38) ~[io.vertx.vertx-core-3.9.1.jar:3.9.1]                  │
│     at io.netty.util.concurrent.AbstractEventExecutor.safeExecute(AbstractEventExecutor.java:164) [io.netty.netty-common-4.1.50.Final.jar:4.1.50.F │
│     at io.netty.util.concurrent.SingleThreadEventExecutor.runAllTasks(SingleThreadEventExecutor.java:472) [io.netty.netty-common-4.1.50.Final.jar: │
│     at io.netty.channel.nio.NioEventLoop.run(NioEventLoop.java:500) [io.netty.netty-transport-4.1.50.Final.jar:4.1.50.Final]                       │
│     at io.netty.util.concurrent.SingleThreadEventExecutor$4.run(SingleThreadEventExecutor.java:989) [io.netty.netty-common-4.1.50.Final.jar:4.1.50 │
│     at io.netty.util.internal.ThreadExecutorMap$2.run(ThreadExecutorMap.java:74) [io.netty.netty-common-4.1.50.Final.jar:4.1.50.Final]             │
│     at io.netty.util.concurrent.FastThreadLocalRunnable.run(FastThreadLocalRunnable.java:30) [io.netty.netty-common-4.1.50.Final.jar:4.1.50.Final] │
│     at java.lang.Thread.run(Thread.java:834) [?:?]

Any ideas on when we might be able to use 2.7.0 with strimzi-kafka-operator?

Cheers

[scholzj]

There is no Kafka 2.7.0 release yet: http://kafka.apache.org/downloads ... the release is still work in progress. Even after it is released, support needs to be first added to Strimzi. It will be supported in the next Strimzi release after the KAfka release. But since it is not clear when it will be released (Kafka releases often go through several RCs to ensure the high quality etc.), I do not dare to speculate.

[feydan]

Does 0.21.1 with KafkaMirrorMaker2 CRD @ 2.7.0 support sync.group.offsets.enabled: true?

I tried adding it to the checkpointConnector config and the flag seems to get set in the logs, but I'm not seeing any groups in my target cluster even though the checkpoints.internal topic is receiving messages.

    mirrors:
    - checkpointConnector:
        config:
          checkpoints.topic.segment.bytes: 1073742
          sync.group.offsets.enabled: true
        tasksMax: 3

[scholzj]

The config section is just a map which is passed to the connector as is. There is no special support for it, but I don't think it should be needed.

[scholzj]

FYI: It worked fine for me in 0.21.1:

    checkpointConnector:
      config:
        sync.group.offsets.enabled: true

The timeouts area bit tricky ... the defaults are 10 minutes to discover new topic / consumer group and the syncing is by default done only once per minute or two. So they make it hard to test sometimes. SO for testing I actually made them much shorter:

    checkpointConnector:
      config:
        refresh.groups.interval.seconds: 60
        sync.group.offsets.enabled: true
        sync.group.offsets.interval.seconds: 1
        emit.checkpoints.interval.seconds: 1

[burkass]

Just want to share, don't forget to commit.close() consumer before move to another cluster because it will take offset auto commit before it. I just got this experience and after I put commit.close() in old cluster, new consumer will consume last offset from old consumer.

Anyway, it was occured when we consumer huge messages in simultaneous