bug: debezium connector failed to initialize after alter connector

[zwang28]

Describe the bug

When rebuilding source reader, 2 debezium connector with the same topic.prefix exist. However, debezium would expect topic.prefix to be unique for each connector. It leads to this connector failure.

Since source reader builder will retry on error, it might resolve the issue automatically.

As a demostration, the connector failure is resolved if we drop the original debezium connector first, i.e.

if should_rebuild_stream {
    stream.replace_data_stream(stream::empty()); -- drop the original connector
    tokio::time::sleep(Duration::from_secs(30)).await; -- wait for the complete drop
    self.rebuild_stream_reader(source_desc, stream)?;
}

However, we cannot rely on the sleep in implementation to ensure the async connector drop completes.

Error message/log

Engine has failed with
...
DebeziumOpenLineageEmitter not initialized for connector
...
Call init() first.

To Reproduce

• create a cdc source.
• alter source xxx connector with (cdc.source.wait.streaming.start.timeout='500');
• reproduced.

Expected behavior

No response

How did you deploy RisingWave?

No response

The version of RisingWave

v2.7

Additional context

No response