[flink] Fix Debezium Avro deserialization for multi topic source

[ArnavBalyan]

Purpose

• KafkaDebeziumAvroDeserializationSchema returns 1 topic via findOneTopic at init time, this gets reused for every incoming record.
• With multi topic configs (topic=a;b), any message coming from another topic leads to a crash with SerializationException: The given schema does not match. The same cached topic is also passed into CdcSourceRecord, misrouting records to the wrong table.
• Read topic from the actual message for deserialize instead of using the cached value at init time.
• Closes user reported bug [Bug] Kafka Debezium Avro deserialization uses wrong schema subject for multi-topic sources #7859.

Tests

• Existing UTs

[ArnavBalyan]

cc @JingsongLi thanks! :)

[ArnavBalyan]

cc @XiaoHongbo-Hope could you PTAL thanks!

[XiaoHongbo-Hope]

cc @XiaoHongbo-Hope could you PTAL thanks!

Sure.

[ArnavBalyan]

thanks!

[ArnavBalyan]

cc @JingsongLi @XiaoHongbo-Hope gentle reminder could PTAL minor bug fix thanks!

[ArnavBalyan]

Gentle reminder thanks!