Hi,
I caught the issue after setting DEBUG for logging mm2 and here is the issue I
see -
[2021-12-03 15:10:20,142] DEBUG [Consumer clientId=consumer-92, groupId=null]
Sending READ_UNCOMMITTED IncrementalFetchRequest(toSend=(bigtopic),
toForget=(), implied=()) to broker sourceip_A:9092 (id: 1002 rack: null)
(org.apache.kafka.clients.consumer.internals.Fetcher:260)
[2021-12-03 15:10:20,143] DEBUG [Producer
clientId=connector-producer-MirrorSourceConnector-81] Exception occurred during
message send: (org.apache.kafka.clients.producer.KafkaProducer:957)
org.apache.kafka.common.errors.RecordTooLargeException: The message is 3850161
bytes when serialized which is larger than 1048576, which is the value of the
max.request.size configuration.
[2021-12-03 15:10:20,143] ERROR WorkerSourceTask{id=MirrorSourceConnector-81}
failed to send record to mongodb.center0.notification:
(org.apache.kafka.connect.runtime.WorkerSourceTask:354)
org.apache.kafka.common.errors.RecordTooLargeException: The message is 3850161
bytes when serialized which is larger than 1048576, which is the value of the
max.request.size configuration.
[2021-12-03 15:10:20,143] DEBUG WorkerSourceTask{id=MirrorSourceConnector-81}
Failed record: SourceRecord{sourcePartition={cluster=, partition=0,
topic=mongodb.center0.notification}, sourceOffset={offset=114777285}}
ConnectRecord{topic='mongodb.center0.notification', kafkaPartition=0,
key=[B@1163c369, keySchema=Schema{BYTES}, value=[B@21b692e9,
valueSchema=Schema{BYTES}, timestamp=1637939789160,
headers=ConnectHeaders(headers=)}
(org.apache.kafka.connect.runtime.WorkerSourceTask:355)
Tried to set the below and restarted the kafka service in target node B where
mm2 is running -
in producer.properties, consumer.properties, server.properties and
mm2.properties
max.request.size=3950161
max.message.bytes=3950161
message.max.bytes=3950161
but no luck!
Could someone please help on the issue as it looks to be a bug again with mm2 -
https://issues.apache.org/jira/browse/KAFKA-6177
Thanks & Regards,
Rijo Roy
On Friday, 3 December, 2021, 07:23:38 pm IST, Rijo Roy
<[email protected]> wrote:
Hi,
After going through this https://issues.apache.org/jira/browse/KAFKA-10582 , I
learnt that this is a similar behaviour seen with Mirrormaker2 and I found some
success initially by restarting the kafka node in cluster B where mm2 instance
was running. Initially the size of the topic increased to 167GB and now
constantly reducing in every minute (size as of now is 39GB). I am stuck in
moving ahead with the same and would really appreciate your help on the same.
Thanks,
Rijo Roy
On Thursday, 2 December, 2021, 11:20:29 pm IST, Rijo Roy <[email protected]>
wrote:
Hi,
Hope you all are safe and well!
I found success in using Mirrormaker2 to sync data from source A (1 broker in a
3node cluster) to target B(1 broker in a 3node cluster) where I was running
mirrormaker2 in the controller node in cluster B.
Now, I see an issue with the sync for a topic whose size in cluster A is
approximately 332GB but the same topic in cluster B is hardly 8KB. And this
issues is not seen for any other topics as they are getting sync'ed as
expected..
Could you please help me understand why this topic is failing to get sync'ed
Source cluster A
=============
kafka-topics --zookeeper zknode:2181 --describe --topic bigtopic
Topic: bigtopic PartitionCount: 1 ReplicationFactor: 3 Configs:
cleanup.policy=delete
Topic: bigtopic Partition: 0 Leader: 1002 Replicas: 1002,1001,1003
Isr: 1003,1002,1001
du -sh /kafkadata/logs/bigtopic-0/
331G /kafkadata/logs/bigtopic-0/
Target cluster B
============
kafka-topics --zookeeper zknodeB:2181 --describe --topic bigtopic
Topic: bigtopic PartitionCount: 1 ReplicationFactor: 3 Configs:
cleanup.policy=delete
Topic: bigtopic Partition: 0 Leader: 6 Replicas: 6,4,5 Isr:
6,4,5
du -sh /kafkadata/logs/bigtopic-0/
8.0K /kafkadata/logs/bigtopic-0/
And here is the mm2.properties that I use-
# mm2.properties
clusters = A, B
A.bootstrap.servers = Ahostname:9092
B.bootstrap.servers = Bhostname:9092
A->B.enabled = true
A->B.topics = somanytopicnames_including_bigtopic
replication.factor=3
refresh.topics.interval.seconds = 10
refresh.groups.interval.seconds = 10
sync.topic.config.enabled = true
replication.policy.separator =
source.cluster.alias =
target.cluster.alias =
#replication.policy.class=
io.strimzi.kafka.connect.mirror.IdentityReplicationPolicy
groups = .*
tasks.max=100
emit.checkpoints.interval.seconds = 10
security.protocol=SASL_PLAINTEXT
sasl.mechanism=PLAIN
sasl.jaas.config=org.apache.kafka.common.security.plain.PlainLoginModule
required \
username="CONFIDENTIAL" \
password="vvvvvvv";
Thanks & Regards,
Rijo Roy
