Bug report

For bug reports, provide this information, please:

What Debezium connector do you use and what version?

DBZ-server 3.02

What is the connector configuration?

1. sink config
   debezium.sink.type=kafka
   debezium.sink.kafka.producer.bootstrap.servers=XXX
   debezium.sink.kafka.producer.key.serializer=org.apache.kafka.common.serialization.StringSerializer
   debezium.sink.kafka.producer.value.serializer=org.apache.kafka.common.serialization.StringSerializer
   debezium.sink.kafka.producer.security.protocol=SASL_SSL
   debezium.sink.kafka.producer.sasl.jaas.config=org.apache.kafka.common.security.plain.PlainLoginModule required username='XXX' password='XXX';
   debezium.sink.kafka.producer.sasl.mechanism=PLAIN
   debezium.sink.kafka.producer.client.dns.lookup=use_all_dns_ips
   debezium.sink.kafka.producer.client=PLAIN
   debezium.sink.kafka.producer.client.id=vmCDC-SX
   debezium.sink.kafka.producer.compression.type=lz4
   debezium.sink.kafka.producer.creation.enable=false

1. source config

debezium.source.connector.class=io.debezium.connector.mysql.MySqlConnector
debezium.source.database.hostname=vmCDC.local
debezium.source.database.port=3306
debezium.source.database.user=cdcUser
debezium.source.database.password=cdcPassword
debezium.source.database.server.id=1
debezium.source.topic.prefix=XXX.test.events
debezium.source.database.include.list=db
debezium.source.column.exclude.list=db.tpltsk_issuestate.IconData
debezium.source.table.include.list=db.rel_country,db.rel_municipality,db.rel_company,db.rel_address,db.rel_relation,db.rel_supplieractivity,db.rel_communicationchannel,db.fin_bankaccount,db.rel_supplieractivitytype,db.sec_user,db.rel_communication,db.rel_financial,db.tpltsk_issuestate,db.tpltsk_issue
debezium.source.offset.storage.file.filename=data/offsets.dat
debezium.source.schema.history.internal.kafka.bootstrap.servers=XXX
debezium.source.schema.history.internal.kafka.sasl.mechanism=PLAIN
debezium.source.schema.history.internal.kafka.topic=XXX.test.events-internal
debezium.source.schema.history.internal.kafka.security.protocol=SASL_SSL
debezium.source.schema.history.internal.kafka.sasl.jaas.config=org.apache.kafka.common.security.plain.PlainLoginModule required username='XXX' password='XXX';
debezium.source.schema.history.internal.consumer.security.protocol=SASL_SSL
debezium.source.schema.history.internal.consumer.sasl.jaas.config=org.apache.kafka.common.security.plain.PlainLoginModule required username='XXX' password='XXX';
debezium.source.schema.history.internal.consumer.sasl.mechanism=PLAIN
debezium.source.schema.history.internal.producer.security.protocol=SASL_SSL
debezium.source.schema.history.internal.producer.sasl.jaas.config=org.apache.kafka.common.security.plain.PlainLoginModule required username='XXX' password='XXX';
debezium.source.schema.history.internal.producer.sasl.mechanism=PLAIN
debezium.source.include.schema.changes=false
debezium.source.topic.creation.default.compression.type=lz4
debezium.source.topic.creation.enable=false
debezium.source.provide.transaction.metadata=true
debezium.source.tombstones.on.delete=false

debezium.source.snapshot.mode=no_data

1. signaling => listens to kafka topic test-signal and executes the commands

debezium.source.signal.enabled.channels=kafka
debezium.source.signal.kafka.topic=XXX.signal
debezium.source.signal.kafka.bootstrap.servers=XXX
debezium.source.signal.kafka.sasl.mechanism=PLAIN
debezium.source.signal.kafka.security.protocol=SASL_SSL
debezium.source.signal.kafka.sasl.jaas.config=org.apache.kafka.common.security.plain.PlainLoginModule required username='XXX' password='XXX';
debezium.source.signal.consumer.security.protocol=SASL_SSL
debezium.source.signal.consumer.sasl.jaas.config=org.apache.kafka.common.security.plain.PlainLoginModule required username='XXX' password='XXX';
debezium.source.signal.consumer.sasl.mechanism=PLAIN
debezium.source.signal.consumer.client.id=signal-listener-for-kafka-agent
debezium.source.signal.producer.security.protocol=SASL_SSL
debezium.source.signal.producer.sasl.jaas.config=org.apache.kafka.common.security.plain.PlainLoginModule required username='XXX' password='XXX';
debezium.source.signal.producer.sasl.mechanism=PLAIN

1. format
   debezium.format.key=json
   debezium.format.key.schemas.enable=false
   debezium.format.value=json
   debezium.format.value.schemas.enable=false
   debezium.format.header=json

1. transformations
   debezium.transforms=beginFilter,endFilter,reroute,tenantInsert
   debezium.transforms.reroute.type=org.apache.kafka.connect.transforms.RegexRouter
   debezium.transforms.reroute.regex=XXX.test.events.([^.]).(.)
   debezium.transforms.reroute.replacement=XXX.test.events
   debezium.transforms.beginFilter.topic.regex=XXX.test.events.transaction
   debezium.transforms.beginFilter.type=io.debezium.transforms.Filter
   debezium.transforms.beginFilter.language=jsr223.groovy
   debezium.transforms.beginFilter.condition=value.status != 'BEGIN'
   debezium.transforms.endFilter.topic.regex=XXX.test.events.transaction
   debezium.transforms.endFilter.type=io.debezium.transforms.Filter
   debezium.transforms.endFilter.language=jsr223.groovy
   debezium.transforms.endFilter.condition=!(value.status == 'END' && value.event_count == 0)
   debezium.transforms.tenantInsert.type=org.apache.kafka.connect.transforms.InsertField$Value
   debezium.transforms.tenantInsert.static.field=tenantId
   debezium.transforms.tenantInsert.static.value=XXX

1. logging
   quarkus.log.console.json=false
   quarkus.log.level=INFO
   quarkus.http.port=8080

   What is the captured database version and mode of deployment?

(E.g. on-premises, with a specific cloud provider, etc.)

on premise mysql 5.6

What behavior do you expect?

Not get an error

What behavior do you see?

90% of the times my snapshot event through signal works but the other 10% i get this error:

Do you see the same behaviour using the latest released Debezium version?

(Ideally, also verify with latest Alpha/Beta/CR version)

I'm on the last major version

Do you have the connector logs, ideally from start till finish?

(You might be asked later to provide DEBUG/TRACE level log)

If needed I can provide extra connector logs then shown on the screenshot