You will include the topic metadata alongside the data that gets passed to the actual serialization process.

Within your custom serialization you will need to access CollectingTableSink or Datastream to where you can manipulate the records before they get serialized. Maybe use a wrapper object that contains the original data row with the metadata

public class MetadataAwareKafkaRecordSerializationSchema 
    extends KafkaDynamicRecordSerializationSchema {

    // constructor to pass necessary configurations
    
    @Override
    public ProducerRecord<byte[], byte[]> serialize(RowData rowData, @Nullable Long timestamp) {
        // Extract topic from metadata (assuming you have a way to access it)
        String topic = ...; // implement logic to fetch topic from rowData or context
        
        // serialize your data (use the super class method or customize based on your needs)
        byte[] data = super.serialize(rowData, timestamp).value();
        
        // create a ProducerRecord that includes the topic metadata
        return new ProducerRecord<>(topic, null, timestamp, null, data);
    }
}