如何从Quarkus消费Debezium主题中的消息？

<dependency>
    <groupId>io.quarkus</groupId>
    <artifactId>quarkus-kafka</artifactId>
</dependency>

import io.quarkus.kafka.client.serialization.JsonbDeserializer;
import io.quarkus.runtime.Startup;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.eclipse.microprofile.reactive.messaging.Incoming;

@Startup
public class MyKafkaConsumer {

    @Incoming("my-topic")
    public void consume(ConsumerRecord<String, MyMessage> record) {
        MyMessage message = record.value();
        // 处理消息
    }
}

kafka.bootstrap.servers=my-kafka-server:9092

mp.messaging.incoming.my-topic.connector=debezium
mp.messaging.incoming.my-topic.debezium.connector.class=io.debezium.connector.mysql.MySqlConnector
mp.messaging.incoming.my-topic.debezium.offset.storage=org.apache.kafka.connect.storage.FileOffsetBackingStore
mp.messaging.incoming.my-topic.debezium.offset.storage.file.filename=data/offsets.dat
mp.messaging.incoming.my-topic.debezium.offset.flush.interval.ms=60000
mp.messaging.incoming.my-topic.debezium.database.hostname=my-db-host
mp.messaging.incoming.my-topic.debezium.database.port=3306
mp.messaging.incoming.my-topic.debezium.database.user=my-db-username
mp.messaging.incoming.my-topic.debezium.database.password=my-db-password
mp.messaging.incoming.my-topic.debezium.database.dbname=my-db-name
mp.messaging.incoming.my-topic.debezium.database.server.id=1
mp.messaging.incoming.my-topic.debezium.database.server.name=my-server-name
mp.messaging.incoming.my-topic.debezium.table.include.list=my-table-name
mp.messaging.incoming.my-topic.debezium.transforms.unwrap.type=io.debezium.transforms.ExtractNewRecordState
mp.messaging.incoming.my-topic.debezium.transforms.unwrap.drop.tombstones=true
mp.messaging.incoming.my-topic.debezium.transforms.unwrap.delete.handling.mode=drop
mp.messaging.incoming.my-topic.value.deserializer=io.quarkus.kafka.client.serialization.JsonbDeserializer
mp.messaging.incoming.my-topic.key.deserializer=org.apache.kafka.common.serialization.StringDeserializer