[FLINK-40102][mysql]Support batch emitting CreateTableEvents to reduce schema coordination overhead

[eric666666]

PR Description

What is the purpose of the change

Add a new option scan.emit.create-table-events.in-batch.enabled for MySQL CDC pipeline connector. When enabled, all CreateTableEvents are collected and emitted in batch during the snapshot initialization phase, significantly reducing schema coordination overhead in scenarios with thousands of tables.

Brief change log

• MySqlDataSourceOptions: Add SCAN_EMIT_CREATE_TABLE_EVENTS_IN_BATCH_ENABLED config option (default false)
• MySqlDataSourceFactory: Read config and pass to MySqlSourceConfigFactory
• MySqlSourceConfig / MySqlSourceConfigFactory: Add emitCreateTableEventsInBatchEnabled field and accessors
• MySqlPipelineRecordEmitter: Add shouldBatchEmitCreateTableEvents() helper; batch-emit all cached CreateTableEvents on first record, populate alreadySendCreateTableTables to prevent duplicate emissions

Verifying this change

• Added MySqlPipelineRecordEmitterTest (6 test cases) covering:
  • Batch emit when enabled
  • Batch emit only once (idempotent)
  • No batch emit when disabled
  • Empty cache no-op
  • Config default propagation
• Updated MySqlDataSourceFactoryTest.testOptionalOption to verify option registration and propagation

[yuxiqian]

IIUC CreateTableEvents are still emitted one-by-one with this option enabled, and lots of schema coordination might block job from starting up.

[eric666666]

Thanks @yuxiqian. The goal of this PR is specifically the INITIAL-mode case: without it, each snapshot split's low-watermark triggers a per-table CreateTableEvent interleaved with the data stream. Each one makes the downstream SchemaOperator emit a FlushEvent, which flushes whatever data is currently buffered — so with thousands of tables the data stream is interrupted thousands of times, and on Paimon/Iceberg that means thousands of tiny files + throughput loss.

With the option on, all CreateTableEvents are emitted together up front (first record, before any DataChangeEvent). The FlushEvent count is unchanged, but those flushes now hit an empty buffer (no data has flowed yet), so no real data files are produced. Per-table JDBC (SHOW CREATE TABLE) is also collapsed from N connections to one batched fetch.

The downstream per-event coordination you pointed at (one blocking RPC per CreateTableEvent) is a separate concern; happy to address it in a follow-up.