MM2 Config does not respect consumer.* parameters

[arunsudhakar]

I use MM2 to replicate data from a source cluster to a freshly instantiated cluster. However, I don't want to replicate all the messages. I only want to replicate messages that have been created after the target cluster has been initiated. Any message created before that is irrelevant. I thought the best way to do this would be to set the auto.offset.reset : latest settings in the config section of the consumer.

This is my CR

apiVersion: kafka.strimzi.io/v1beta2
kind: KafkaMirrorMaker2
metadata:
  name: test-mirror-maker-2
spec:
  version: 2.8.0
  replicas: 1
  #target cluster
  connectCluster: "test"
  clusters:
  #Consumer
  - alias: "datalake"
    bootstrapServers: datalake-kafka-0.datalake-kafka-brokers.dataflow-pipeline.svc:9092
    config:
      consumer.auto.offset.reset: latest
      consumer.metrics.recording.level: DEBUG
  #Producer
  - alias: "test"
    bootstrapServers: test-kafka-brokers.atwin-sessionid.svc:9092
    config:
      producer.max.request.size: 11111111 
      producer.metrics.recording.level: DEBUG
      config.storage.replication.factor: -1
      offset.storage.replication.factor: -1
      status.storage.replication.factor: -1
  mirrors:
  - sourceCluster: "datalake"
    targetCluster: "test"
    sourceConnector:
      config:
        replication.factor: 1
        offset-syncs.topic.replication.factor: 1
        sync.topic.acls.enabled: "false"
    heartbeatConnector:
      config:
        heartbeats.topic.replication.factor: 1
    checkpointConnector:
      config:
        checkpoints.topic.replication.factor: 1
    topicsPattern: ".*"
    groupsPattern: ".*"

In the logs, for the ConsumerConfig values, it still reads as

2021-10-18T13:34:44.692309499Z 2021-10-18 13:34:44,692 INFO ConsumerConfig values: 
2021-10-18T13:34:44.692333499Z 	allow.auto.create.topics = true
2021-10-18T13:34:44.692358499Z 	auto.commit.interval.ms = 5000
2021-10-18T13:34:44.692364299Z 	auto.offset.reset = earliest

but the ProducerConfig section respects the values given in the config

2021-10-18T13:34:43.708493339Z 2021-10-18 13:34:43,708 INFO ProducerConfig values: 
2021-10-18T13:34:43.708514439Z 	acks = -1
2021-10-18T13:34:43.708520339Z 	batch.size = 16384
2021-10-18T13:34:43.708546239Z 	bootstrap.servers = [kafka-kafka-brokers.atwin-sessionid.svc:9092]
2021-10-18T13:34:43.708552339Z 	buffer.memory = 33554432
2021-10-18T13:34:43.708558640Z 	client.dns.lookup = use_all_dns_ips
...
...
2021-10-18T13:34:43.708605140Z 	max.block.ms = 9223372036854775807
2021-10-18T13:34:43.708634240Z 	max.in.flight.requests.per.connection = 1
2021-10-18T13:34:43.708639140Z 	max.request.size = 11111111
2021-10-18T13:34:43.708661840Z 	metrics.recording.level = DEBUG

I'm using Strimzi v 0.26 with Kafka Version 2.8.0

[scholzj]

Kafka Connect creates many different consumers and producers. By cutting log snippets like this, it is not really clear what exactly are you sharing I'm afraid.

[arunsudhakar]

Sorry, I should have been more specific by attaching the full mm2.log and the manifest.
I've REDACTED the topic names for confidentiality reasons

Lines 29 to 31 in the YAML file has the parameters I'm passing to the deployment and they can be seen in the log files on lines 41,42 and 45

The Producer config for max.request.size can be seen as accepted in lines 1963, 3155, 3270, 4882, 5305, 5378

However for the consumer.auto.offset.reset, the value is still set to earliest as can be seen on lines 1106, 1372, 1595, 2242, 3551, 5557 & 5853

To reiterate what I'm trying to do -
I've 1 Main Kafka cluster A, which has data constantly being pumped into it. I create another cluster B, for a temporary session and that cluster should only replicate data from A which has been created only after cluster B has been online and nothing before. Hence the latest offset. Shouldn't what I was trying to do the MM2 config above help achieve this? If not, what else can I do? I cannot use consumer commits, because there might not be active consumers for a long period of time.

[arunsudhakar]

From what I understand, it does as is evident in the src for MM2 Connector Config ,

Line 262 is props.putIfAbsent(AUTO_OFFSET_RESET_CONFIG, "earliest");

defaulted to earliest.

This file is referenced under https://kafka.apache.org/30/documentation/#georeplication-mirrormaker

[scholzj]

So, maybe you need to put it into the Source connector configuration? Maybe you should not use the consumer prefix? I guess you will beed to try it out.

[arunsudhakar]

I put it into the Source connector configuration and managed to get the value set to latest , but it still was copying from the beginning. There might be something else I need to change, but not sure what. So, for now I'm using MirrorMaker where the default behaviour matches my expectation. I'll come back to MM2 later on and try to figure out what the issue is. Thanks for your help

[scholzj]

Keep in mind that the option is used only when there is no previous offset commit yet. So changing it around without getting rid of the old commits will not change anything.

[vgulch]

i'm having the same issue. i cant manage to set 'auto.offset.reset = earliest' for consumer.
I was trying with kafka operator v0.20.0 and v0.27.1
@arunsudhakar did you manage to make it work?

I put it into the Source connector configuration and managed to get the value set to latest
how did you configured it, i was trying many things but its always set 'earliest' as a value